Kafka update to 3.5.1

.gitignore

@@ -6,3 +6,4 @@ target/
 .project
 *.DS_Store
 *.log
+devicehive-common/dependency-reduced-pom.xml

devicehive-auth/src/main/java/com/devicehive/application/AppContextEventListener.java

@@ -0,0 +1,70 @@
+package com.devicehive.application;
+
+/*
+ * #%L
+ * DeviceHive Frontend Logic
+ * %%
+ * Copyright (C) 2016 DataArt
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.springframework.boot.context.event.ApplicationPreparedEvent;
+import org.springframework.boot.context.event.SpringApplicationEvent;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ApplicationContextEvent;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.MapPropertySource;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+public class AppContextEventListener implements ApplicationListener<ApplicationEvent> {
+
+    private void printActiveProperties(ConfigurableEnvironment env) {
+
+        System.out.println("************************* ACTIVE APP PROPERTIES ******************************");
+
+        List<MapPropertySource> propertySources = new ArrayList<>();
+
+        env.getPropertySources().forEach(it -> {
+            if (it instanceof MapPropertySource) {
+                propertySources.add((MapPropertySource) it);
+            }
+        });
+
+        propertySources.stream()
+                .map(propertySource -> propertySource.getSource().keySet())
+                .flatMap(Collection::stream)
+                .distinct()
+                .sorted()
+                .forEach(key -> {
+                    try {
+                        System.out.println(key + "=" + env.getProperty(key));
+                    } catch (Exception e) {
+                        System.out.println(e);
+                    }
+                });
+        System.out.println("******************************************************************************");
+    }
+
+    @Override
+    public void onApplicationEvent(ApplicationEvent event) {
+        if(event instanceof ApplicationPreparedEvent) {
+            printActiveProperties((ConfigurableEnvironment)  ((ApplicationPreparedEvent) event).getApplicationContext().getEnvironment());
+        }
+    }
+}

devicehive-auth/src/main/java/com/devicehive/application/AuthProxyClientConfig.java

@@ -21,7 +21,7 @@
  */
 
 import com.devicehive.api.RequestResponseMatcher;
-import com.devicehive.proxy.AuthProxyClient;
+import com.devicehive.proxy.AuthRpcClient;
 import com.devicehive.proxy.ProxyResponseHandler;
 import com.devicehive.proxy.api.NotificationHandler;
 import com.devicehive.proxy.client.WebSocketKafkaProxyClient;
@@ -59,10 +59,10 @@ public NotificationHandler notificationHandler(Gson gson, RequestResponseMatcher
     }
 
     @Bean
-    public AuthProxyClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson) {
+    public AuthRpcClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson) {
         WebSocketKafkaProxyClient proxyClient = new WebSocketKafkaProxyClient(notificationHandler);
         proxyClient.setWebSocketKafkaProxyConfig(proxyConfig);
-        AuthProxyClient client = new AuthProxyClient(REQUEST_TOPIC, responseTopic, proxyClient, requestResponseMatcher, gson);
+        AuthRpcClient client = new AuthRpcClient(REQUEST_TOPIC, responseTopic, proxyClient, requestResponseMatcher, gson);
         client.start();
         return client;
     }

devicehive-auth/src/main/java/com/devicehive/application/DeviceHiveAuthApplication.java

@@ -57,6 +57,7 @@ public class DeviceHiveAuthApplication extends SpringBootServletInitializer {
 
     public static void main(String... args) {
         ConfigurableApplicationContext context = new SpringApplicationBuilder()
+                .listeners(new AppContextEventListener())
                 .sources(DeviceHiveAuthApplication.class)
                 .web(WebApplicationType.SERVLET)
                 .run(args);

devicehive-auth/src/main/java/com/devicehive/proxy/AuthProxyClient.java → devicehive-auth/src/main/java/com/devicehive/proxy/AuthRpcClient.java

@@ -44,16 +44,16 @@
 import java.util.function.Consumer;
 
 @Profile("ws-kafka-proxy")
-public class AuthProxyClient implements RpcClient {
-    private static final Logger logger = LoggerFactory.getLogger(AuthProxyClient.class);
+public class AuthRpcClient implements RpcClient {
+    private static final Logger logger = LoggerFactory.getLogger(AuthRpcClient.class);
 
     private final String requestTopic;
     private final String replyToTopic;
     private final ProxyClient client;
     private final RequestResponseMatcher requestResponseMatcher;
     private final Gson gson;
 
-    public AuthProxyClient(String requestTopic, String replyToTopic, ProxyClient client, RequestResponseMatcher requestResponseMatcher, Gson gson) {
+    public AuthRpcClient(String requestTopic, String replyToTopic, ProxyClient client, RequestResponseMatcher requestResponseMatcher, Gson gson) {
         this.requestTopic = requestTopic;
         this.replyToTopic = replyToTopic;
         this.client = client;
@@ -84,8 +84,8 @@ public void start() {
         client.start();
         createTopic(Arrays.asList(requestTopic, replyToTopic));
         subscribeToTopic(replyToTopic);
-
         pingServer();
+
     }
 
     public void createTopic(List<String> topics) {
@@ -110,7 +110,8 @@ private void pingServer() {
             requestResponseMatcher.addRequestCallback(request.getCorrelationId(), pingFuture::complete);
             logger.debug("Request callback added for request: {}, correlationId: {}", request.getBody(), request.getCorrelationId());
 
-            client.push(ProxyMessageBuilder.notification(new NotificationCreatePayload(requestTopic, gson.toJson(request)))); // toDo: use request partition key
+            client.push(ProxyMessageBuilder.notification(
+                    new NotificationCreatePayload(requestTopic, gson.toJson(request), request.getPartitionKey())));
 
             Response response = null;
             try {

devicehive-auth/src/main/resources/application.properties

@@ -52,6 +52,6 @@ jwt.refresh-token-max-age=15724800000
 jwt.access-token-max-age=1800000
 
 
-spring.flyway.baselineOnMigrate=false
+spring.flyway.baselineOnMigrate=true
 spring.flyway.table=schema_version
 #spring.flyway.baseline-version=3.4.12

devicehive-backend/src/test/java/com/devicehive/base/AbstractSpringTest.java

@@ -75,7 +75,7 @@ public abstract class AbstractSpringTest {
     }
 
     @ClassRule
-    public static KafkaEmbeddedRule kafkaRule = new KafkaEmbeddedRule(true, 1, REQUEST_TOPIC, RESPONSE_TOPIC);
+    public static KafkaEmbeddedRule kafkaRule = new KafkaEmbeddedRule();
 
     @Rule
     public Timeout testTimeout = new Timeout(60000, TimeUnit.MILLISECONDS); // 60k ms = 1 minute

devicehive-frontend/src/test/java/com/devicehive/base/AbstractSpringKafkaTest.java

@@ -50,7 +50,7 @@ public abstract class AbstractSpringKafkaTest {
     }
 
     @ClassRule
-    public static KafkaEmbeddedRule kafkaRule = new KafkaEmbeddedRule(true, 5, REQUEST_TOPIC, RESPONSE_TOPIC);
+    public static KafkaEmbeddedRule kafkaRule = new KafkaEmbeddedRule();
 
     @Rule
     public Timeout testTimeout = new Timeout(180000, TimeUnit.MILLISECONDS); // 180k ms = 3 minutes

devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/FrontendProxyClient.java → devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/FrontendRpcClient.java

@@ -23,7 +23,6 @@
 import com.devicehive.api.RequestResponseMatcher;
 import com.devicehive.model.ServerEvent;
 import com.devicehive.proxy.api.NotificationHandler;
-import com.devicehive.proxy.api.ProxyClient;
 import com.devicehive.proxy.api.ProxyMessageBuilder;
 import com.devicehive.proxy.api.payload.NotificationCreatePayload;
 import com.devicehive.proxy.api.payload.SubscribePayload;
@@ -44,8 +43,8 @@
 import java.util.concurrent.*;
 import java.util.function.Consumer;
 
-public class FrontendProxyClient implements RpcClient {
-    private static final Logger logger = LoggerFactory.getLogger(FrontendProxyClient.class);
+public class FrontendRpcClient implements RpcClient {
+    private static final Logger logger = LoggerFactory.getLogger(FrontendRpcClient.class);
 
     private final String requestTopic;
     private final String replyToTopic;
@@ -56,7 +55,7 @@ public class FrontendProxyClient implements RpcClient {
     private final Gson gson;
     private final RingBuffer<ServerEvent> ringBuffer;
 
-    public FrontendProxyClient(String requestTopic, String replyToTopic, WebSocketKafkaProxyConfig proxyConfig, NotificationHandler notificationHandler, RequestResponseMatcher requestResponseMatcher, Gson gson, RingBuffer<ServerEvent> ringBuffer) {
+    public FrontendRpcClient(String requestTopic, String replyToTopic, WebSocketKafkaProxyConfig proxyConfig, NotificationHandler notificationHandler, RequestResponseMatcher requestResponseMatcher, Gson gson, RingBuffer<ServerEvent> ringBuffer) {
         this.requestTopic = requestTopic;
         this.replyToTopic = replyToTopic;
         this.proxyConfig = proxyConfig;
@@ -118,7 +117,7 @@ private void pingServer() {
         boolean connected = false;
         int attempts = 10;
         for (int i = 0; i < attempts; i++) {
-            logger.info("Ping WebSocket Proxy attempt {}", i);
+            logger.info("Ping WebSocket Proxy Server attempt {}", i);
 
             CompletableFuture<Response> pingFuture = new CompletableFuture<>();
 
@@ -132,9 +131,9 @@ private void pingServer() {
             try {
                 response = pingFuture.get(3000, TimeUnit.MILLISECONDS);
             } catch (InterruptedException | ExecutionException e) {
-                logger.error("Exception occured while trying to ping Backend Server ", e);
+                logger.error("Exception occured while trying to ping WebSocket Proxy Server ", e);
             } catch (TimeoutException e) {
-                logger.warn("Backend Server didn't respond to ping request");
+                logger.warn("WebSocket Proxy Server didn't respond to ping request");
                 continue;
             } finally {
                 requestResponseMatcher.removeRequestCallback(request.getCorrelationId());
@@ -147,10 +146,10 @@ private void pingServer() {
             }
         }
         if (connected) {
-            logger.info("Successfully connected to Backend Server");
+            logger.info("Successfully connected to WebSocket Proxy Server");
         } else {
-            logger.error("Unable to reach out Backend Server in {} attempts", attempts);
-            throw new RuntimeException("Backend Server is not reachable");
+            logger.error("Unable to reach out WebSocket Proxy Server in {} attempts", attempts);
+            throw new RuntimeException("WebSocket Proxy Server is not reachable");
         }
     }
 }

devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/config/FrontendProxyClientConfig.java

@@ -22,7 +22,7 @@
 
 import com.devicehive.api.RequestResponseMatcher;
 import com.devicehive.model.ServerEvent;
-import com.devicehive.proxy.FrontendProxyClient;
+import com.devicehive.proxy.FrontendRpcClient;
 import com.devicehive.proxy.ProxyResponseHandler;
 import com.devicehive.proxy.api.NotificationHandler;
 import com.devicehive.shim.api.client.RpcClient;
@@ -97,7 +97,7 @@ public WorkerPool<ServerEvent> workerPool(Gson gson, RequestResponseMatcher requ
     public RpcClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson, WorkerPool<ServerEvent> workerPool) {
         final ExecutorService execService = Executors.newFixedThreadPool(proxyConfig.getWorkerThreads());
         RingBuffer<ServerEvent> ringBuffer = workerPool.start(execService);
-        RpcClient client = new FrontendProxyClient(REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, notificationHandler, requestResponseMatcher, gson, ringBuffer);
+        RpcClient client = new FrontendRpcClient(REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, notificationHandler, requestResponseMatcher, gson, ringBuffer);
         client.start();
         return client;
     }

devicehive-rdbms-dao/pom.xml

@@ -16,7 +16,7 @@
 
     <properties>
         <project.rootdir>${project.parent.basedir}</project.rootdir>
-        <redisson.version>3.17.7</redisson.version>
+        <redisson.version>3.23.5</redisson.version>
     </properties>
 
     <dependencies>
@@ -56,16 +56,16 @@
             <groupId>org.hibernate.validator</groupId>
             <artifactId>hibernate-validator</artifactId>
         </dependency>
-        <dependency>
-            <groupId>javax.el</groupId>
-            <artifactId>javax.el-api</artifactId>
-            <version>${javax.el.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.glassfish.web</groupId>
-            <artifactId>javax.el</artifactId>
-            <version>${javax.el.version}</version>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>javax.el</groupId>-->
+<!--            <artifactId>javax.el-api</artifactId>-->
+<!--            <version>${javax.el.version}</version>-->
+<!--        </dependency>-->
+<!--        <dependency>-->
+<!--            <groupId>org.glassfish.web</groupId>-->
+<!--            <artifactId>javax.el</artifactId>-->
+<!--            <version>${javax.el.version}</version>-->
+<!--        </dependency>-->
     </dependencies>
 
 </project>

devicehive-rdbms-dao/src/main/resources/db/migration/V3_0_10__drop_network_auto_creation.sql

@@ -2,7 +2,7 @@
 -- #%L
 -- DeviceHive Dao RDBMS Implementation
 -- %%
--- Copyright (C) 2016-2017 DataArt
+-- Copyright (C) 2016 DataArt
 -- %%
 -- Licensed under the Apache License, Version 2.0 (the "License");
 -- you may not use this file except in compliance with the License.

devicehive-rdbms-dao/src/main/resources/redisson.yaml

@@ -2,6 +2,7 @@ codec:
   class: org.redisson.codec.SerializationCodec
 singleServerConfig:
   address: "redis://${REDIS_MASTER_HOST}:${REDIS_MASTER_PORT}"
+  password: "${REDIS_MASTER_PASSWORD}"
   idleConnectionTimeout: 10000
   connectTimeout: 10000
   timeout: 3000

devicehive-shim-kafka-impl/src/main/java/com/devicehive/shim/config/KafkaRpcConfig.java

@@ -149,4 +149,8 @@ public int getConnectionTimeout() {
     public int getHandlerThreads() {
         return handlerThreads;
     }
+
+    public String getBootstrapServers() {
+        return bootstrapServers;
+    }
 }