MINOR: Modify KafkaEventQueue VoidEvent to as singleton and use more proper function interface (#19356)

class `VoidEvent` provides singleton object , but nobody use it.  I
think we should private `VoidEvent` constructor and only use singleton.
use `UnaryOperator<OptionalLong>` instead
`Function<OptionalLong,OptionalLong>`

Reviewers: Chia-Ping Tsai <[email protected]>

Author: gongxuanzhang

Diff for: server-common/src/main/java/org/apache/kafka/queue/EventQueue.java

@@ -17,11 +17,9 @@
 
 package org.apache.kafka.queue;
 
-import org.slf4j.Logger;
 
 import java.util.OptionalLong;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 
 public interface EventQueue extends AutoCloseable {
@@ -44,38 +42,20 @@ interface Event {
         default void handleException(Throwable e) {}
     }
 
-    abstract class FailureLoggingEvent implements Event {
-        private final Logger log;
-
-        public FailureLoggingEvent(Logger log) {
-            this.log = log;
-        }
-
-        @Override
-        public void handleException(Throwable e) {
-            if (e instanceof RejectedExecutionException) {
-                log.info("Not processing {} because the event queue is closed.", this);
-            } else {
-                log.error("Unexpected error handling {}", this, e);
-            }
-        }
-
-        @Override
-        public String toString() {
-            return this.getClass().getSimpleName();
-        }
-    }
-
-    class NoDeadlineFunction implements Function<OptionalLong, OptionalLong> {
+    class NoDeadlineFunction implements UnaryOperator<OptionalLong> {
         public static final NoDeadlineFunction INSTANCE = new NoDeadlineFunction();
+        
+        private NoDeadlineFunction() {
+            
+        }
 
         @Override
         public OptionalLong apply(OptionalLong ignored) {
             return OptionalLong.empty();
         }
     }
 
-    class DeadlineFunction implements Function<OptionalLong, OptionalLong> {
+    class DeadlineFunction implements UnaryOperator<OptionalLong> {
         private final long deadlineNs;
 
         public DeadlineFunction(long deadlineNs) {
@@ -88,7 +68,7 @@ public OptionalLong apply(OptionalLong ignored) {
         }
     }
 
-    class EarliestDeadlineFunction implements Function<OptionalLong, OptionalLong> {
+    class EarliestDeadlineFunction implements UnaryOperator<OptionalLong> {
         private final long newDeadlineNs;
 
         public EarliestDeadlineFunction(long newDeadlineNs) {
@@ -107,28 +87,13 @@ public OptionalLong apply(OptionalLong prevDeadlineNs) {
         }
     }
 
-    class LatestDeadlineFunction implements Function<OptionalLong, OptionalLong> {
-        private final long newDeadlineNs;
-
-        public LatestDeadlineFunction(long newDeadlineNs) {
-            this.newDeadlineNs = newDeadlineNs;
-        }
-
-        @Override
-        public OptionalLong apply(OptionalLong prevDeadlineNs) {
-            if (prevDeadlineNs.isEmpty()) {
-                return OptionalLong.of(newDeadlineNs);
-            } else if (prevDeadlineNs.getAsLong() > newDeadlineNs) {
-                return prevDeadlineNs;
-            } else {
-                return OptionalLong.of(newDeadlineNs);
-            }
-        }
-    }
-
     class VoidEvent implements Event {
         public static final VoidEvent INSTANCE = new VoidEvent();
-
+        
+        private VoidEvent() {
+            
+        }
+        
         @Override
         public void run() throws Exception {
         }
@@ -182,7 +147,7 @@ default void appendWithDeadline(long deadlineNs, Event event) {
      * @param event                 The event to schedule.
      */
     default void scheduleDeferred(String tag,
-                                  Function<OptionalLong, OptionalLong> deadlineNsCalculator,
+                                  UnaryOperator<OptionalLong> deadlineNsCalculator,
                                   Event event) {
         enqueue(EventInsertionType.DEFERRED, tag, deadlineNsCalculator, event);
     }
@@ -224,7 +189,7 @@ enum EventInsertionType {
      */
     void enqueue(EventInsertionType insertionType,
                  String tag,
-                 Function<OptionalLong, OptionalLong> deadlineNsCalculator,
+                 UnaryOperator<OptionalLong> deadlineNsCalculator,
                  Event event);
 
     /**

Diff for: server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 
 public final class KafkaEventQueue implements EventQueue {
@@ -444,7 +445,7 @@ public KafkaEventQueue(
         LogContext logContext,
         String threadNamePrefix
     ) {
-        this(time, logContext, threadNamePrefix, VoidEvent::new);
+        this(time, logContext, threadNamePrefix, VoidEvent.INSTANCE);
     }
 
     public KafkaEventQueue(
@@ -472,7 +473,7 @@ public Time time() {
     @Override
     public void enqueue(EventInsertionType insertionType,
                         String tag,
-                        Function<OptionalLong, OptionalLong> deadlineNsCalculator,
+                        UnaryOperator<OptionalLong> deadlineNsCalculator,
                         Event event) {
         EventContext eventContext = new EventContext(event, insertionType, tag);
         Exception e = eventHandler.enqueue(eventContext, deadlineNsCalculator);

Diff for: server/src/main/java/org/apache/kafka/server/AssignmentsManagerDeadlineFunction.java

@@ -20,14 +20,14 @@
 import org.apache.kafka.common.utils.ExponentialBackoff;
 
 import java.util.OptionalLong;
-import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 import static org.apache.kafka.common.requests.AssignReplicasToDirsRequest.MAX_ASSIGNMENTS_PER_REQUEST;
 
 /**
  * This class calculates when the MaybeSendAssignmentsEvent should run for AssignmentsManager.
  */
-public class AssignmentsManagerDeadlineFunction implements Function<OptionalLong, OptionalLong> {
+public class AssignmentsManagerDeadlineFunction implements UnaryOperator<OptionalLong> {
 
     /**
      * The exponential backoff to use.