improved IPC server logging

Author: jlangch

src/main/java/com/github/jlangch/venice/impl/threadpool/ManagedExecutor.java

@@ -57,18 +57,21 @@ public final boolean isShutdown() {
         }
     }
 
-    public final void awaitTermination(final long timeoutMillis) {
+    public final boolean awaitTermination(final long timeoutMillis) {
         synchronized(this) {
             if (executor != null) {
                 try {
-                    executor.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS);
+                    return executor.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS);
                 }
                 catch(Exception ex) {
                     throw new VncException(
                             "Failed awaiting for executor termination",
                             ex);
                 }
             }
+            else {
+                return true;
+            }
         }
     }
 

src/main/java/com/github/jlangch/venice/util/ipc/TcpServer.java

@@ -405,15 +405,27 @@ public void start(final Function<IMessage,IMessage> handler) {
     @Override
     public void close() throws IOException {
         if (started.compareAndSet(true, false)) {
-            final String msg = "Closed server on port " + port + "!";
-            logger.info("server", msg);
+            logger.info("server", "Server closing...");
 
             // do not shutdown the thread-pools too early
             IO.sleep(300);
 
             safeClose(server.get());
             server.set(null);
+
+            // This method does not wait for actively executing tasks to terminate.
             mngdExecutor.shutdownNow();
+
+            // wait max 1'000ms
+            final boolean terminated = mngdExecutor.awaitTermination(1_000);
+
+            IO.sleep(100);
+
+            logger.info(
+                "server",
+                terminated
+                    ? "Server closed."
+                    : "Server closed. Some connection are delaying shutdown confirmation.");
         }
     }
 
@@ -539,14 +551,21 @@ public boolean existsQueue(final String queueName) {
     private void safeClose(final ServerSocketChannel ch) {
         if (ch != null) {
             try {
-                // close the durable queues
-                if (wal.isEnabled()) {
-                    wal.close(p2pQueues.values());
+                try {
+                    // close the durable queues
+                    if (wal.isEnabled()) {
+                        wal.close(p2pQueues.values());
+                    }
+                }
+                catch(Exception ex) {
+                    logger.warn("server", "Error while closing queue WALs.", ex);
                 }
 
                 ch.close();
             }
-            catch(Exception ignore) {}
+            catch(Exception ex) {
+                logger.warn("server", "Error while closing server.", ex);
+            }
         }
     }
 

src/main/java/com/github/jlangch/venice/util/ipc/impl/TcpServerConnection.java

@@ -419,6 +419,10 @@ private Message handleSubscribe(final Message request) {
         // register subscription
         subscriptions.addSubscriptions(request.getTopicsSet(), this);
 
+        logger.info(
+                "conn-" + connectionId,
+                String.format("Subscribed to topics: %s.", Topics.encode(request.getTopics())));
+
         // acknowledge the subscription
         return createOkTextMessageResponse(request, "Subscribed to the topics.");
     }
@@ -427,6 +431,10 @@ private Message handleUnsubscribe(final Message request) {
         // unregister subscription
         subscriptions.removeSubscriptions(request.getTopicsSet(), this);
 
+        logger.info(
+                "conn-" + connectionId,
+                String.format("Unsubscribed from topics: %s.", Topics.encode(request.getTopics())));
+
         // acknowledge the unsubscription
         return createOkTextMessageResponse(request, "Unsubscribed from the topics.");
     }
@@ -1024,16 +1032,18 @@ private void auditPublishError(
             errorBuffer.offer(new Error(errorMsg, msg, ex));
             statistics.incrementDiscardedPublishCount();
         }
-        catch(InterruptedException ignore) {}
+        catch(InterruptedException ignore) { }
     }
 
     private void removeAllChannelTemporaryQueues() {
         try {
             final Set<String> names = tmpQueues.keySet();
             names.forEach(n -> p2pQueues.remove(n));
             tmpQueues.clear();
+
+            logger.info("conn-" + connectionId, "Removed all temporary queues of the connnection");
         }
-        catch(Exception ignore) {}
+        catch(Exception ignore) { }
     }
 
     private long countStandardQueues() {