MINOR: Various code cleanups in server (#22021)

Small Java cleanups fixing logging and types

Reviewers: Andrew Schofield <aschofield@confluent.io>

Author: mimaison

server/src/main/java/org/apache/kafka/server/BrokerLifecycleManager.java

@@ -570,10 +570,10 @@ public void run() {
                 return;
             }
             if (response.authenticationException() != null) {
-                logger.error("Unable to register broker $nodeId because of an authentication exception.", response.authenticationException());
+                logger.error("Unable to register broker {} because of an authentication exception.", nodeId, response.authenticationException());
                 scheduleNextCommunicationAfterFailure();
             } else if (response.versionMismatch() != null) {
-                logger.error("Unable to register broker $nodeId because of an API version problem.", response.versionMismatch());
+                logger.error("Unable to register broker {} because of an API version problem.", nodeId, response.versionMismatch());
                 scheduleNextCommunicationAfterFailure();
             } else if (response.responseBody() == null) {
                 logger.warn("Unable to register broker {}.", nodeId);

server/src/main/java/org/apache/kafka/server/DelayedActionQueue.java

@@ -43,7 +43,7 @@ public void tryCompleteActions() {
                 Runnable action = queue.poll();
                 if (action == null) return;
                 action.run();
-            } catch (Throwable e) {
+            } catch (Exception e) {
                 LOGGER.error("failed to complete delayed actions", e);
             }
         }

server/src/main/java/org/apache/kafka/server/share/session/ShareSession.java

@@ -89,7 +89,7 @@ public synchronized int size() {
         return partitionMap.size();
     }
 
-    public synchronized Boolean isEmpty() {
+    public synchronized boolean isEmpty() {
         return partitionMap.isEmpty();
     }
 
@@ -137,11 +137,11 @@ public synchronized int updateCachedSize() {
         return previousSize != -1 ? cachedSize - previousSize : cachedSize;
     }
 
-    public static String partitionsToLogString(Collection<TopicIdPartition> partitions, Boolean traceEnabled) {
+    public static String partitionsToLogString(Collection<TopicIdPartition> partitions, boolean traceEnabled) {
         if (traceEnabled) {
-            return String.format("( %s )", String.join(", ", partitions.toString()));
+            return partitions.toString();
         }
-        return String.format("%s partition(s)", partitions.size());
+        return partitions.size() + " partition(s)";
     }
 
     public String toString() {

server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java

@@ -93,10 +93,10 @@ public ShareSessionCache(int maxEntries) {
      * Get a session by session key.
      *
      * @param key The share session key.
-     * @return The session, or None if no such session was found.
+     * @return The session, or null if no such session was found.
      */
     public synchronized ShareSession get(ShareSessionKey key) {
-        return sessions.getOrDefault(key, null);
+        return sessions.get(key);
     }
 
     /**

server/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java

@@ -39,7 +39,7 @@ public void testPartitionsToLogString() {
         assertEquals("2 partition(s)", response);
 
         response = ShareSession.partitionsToLogString(partitions, true);
-        assertEquals(String.format("( [%s:foo-0, %s:bar-1] )", uuid1, uuid2), response);
+        assertEquals(String.format("[%s:foo-0, %s:bar-1]", uuid1, uuid2), response);
     }
 
     @Test
@@ -48,6 +48,6 @@ public void testPartitionsToLogStringEmpty() {
         assertEquals("0 partition(s)", response);
 
         response = ShareSession.partitionsToLogString(List.of(), true);
-        assertEquals("( [] )", response);
+        assertEquals("[]", response);
     }
 }