Renamed two classes, changed ping server to be the same for broth Rpc clients, moved kafka version back to 3.2.0

Author: Aleksandar Osenov

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/proxy/AuthProxyClient.java renamed to 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;
@@ -82,9 +82,9 @@ public void push(Request request) {
     @Override
     public void start() {
         client.start();
-        pingServer();
         createTopic(Arrays.asList(requestTopic, replyToTopic));
         subscribeToTopic(replyToTopic);
+        pingServer();
 
     }
 
@@ -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-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/FrontendProxyClient.java renamed to 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;
     }

dockerfiles/devicehive-kafka.Dockerfile

@@ -1,4 +1,4 @@
-FROM docker.io/bitnami/kafka:3.5.1
+FROM docker.io/bitnami/kafka:3.2.0
 
 MAINTAINER devicehive
 

pom.xml

@@ -41,7 +41,7 @@
         <javax.ws.rs-api.version>2.1.1</javax.ws.rs-api.version>
         <javax.websocket.version>1.1</javax.websocket.version>
         <javax.el.version>2.2.4</javax.el.version>
-        <kafka.version>3.5.1</kafka.version>
+        <kafka.version>3.2.0</kafka.version>
         <scala-binaries.version>2.13</scala-binaries.version>
         <slf4j.version>1.7.5</slf4j.version>
         <logback.version>1.1.3</logback.version>