MINOR: Switch anonymous classes to lambda expressions in tools module

tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java

@@ -261,65 +261,68 @@ void testAdminClient() throws Throwable {
                     nodes.size(), testConfig.numClusterNodes);
             }
             tryFeature("createTopics", testConfig.createTopicsSupported,
-                new Invoker() {
-                    @Override
-                    public void invoke() throws Throwable {
-                        try {
-                            client.createTopics(Collections.singleton(
-                                new NewTopic("newtopic", 1, (short) 1))).all().get();
-                        } catch (ExecutionException e) {
-                            throw e.getCause();
-                        }
+                () -> {
+                    try {
+                        client.createTopics(Collections.singleton(
+                            new NewTopic("newtopic", 1, (short) 1))).all().get();
+                    } catch (ExecutionException e) {
+                        throw e.getCause();
                     }
                 },
-                new ResultTester() {
-                    @Override
-                    public void test() throws Throwable {
-                        while (true) {
-                            try {
-                                client.describeTopics(Collections.singleton("newtopic")).all().get();
-                                break;
-                            } catch (ExecutionException e) {
-                                if (e.getCause() instanceof UnknownTopicOrPartitionException)
-                                    continue;
-                                throw e;
-                            }
-                        }
-                    }
-                });
+                () ->  createTopicsResultTest(client, Collections.singleton("newtopic"))
+            );
+
             while (true) {
                 Collection<TopicListing> listings = client.listTopics().listings().get();
                 if (!testConfig.createTopicsSupported)
                     break;
-                boolean foundNewTopic = false;
-                for (TopicListing listing : listings) {
-                    if (listing.name().equals("newtopic")) {
-                        if (listing.isInternal())
-                            throw new KafkaException("Did not expect newtopic to be an internal topic.");
-                        foundNewTopic = true;
-                    }
-                }
-                if (foundNewTopic)
+
+                if (topicExists(listings, "newtopic"))
                     break;
+
                 Thread.sleep(1);
                 log.info("Did not see newtopic.  Retrying listTopics...");
             }
+
             tryFeature("describeAclsSupported", testConfig.describeAclsSupported,
-                new Invoker() {
-                    @Override
-                    public void invoke() throws Throwable {
-                        try {
-                            client.describeAcls(AclBindingFilter.ANY).values().get();
-                        } catch (ExecutionException e) {
-                            if (e.getCause() instanceof SecurityDisabledException)
-                                return;
-                            throw e.getCause();
-                        }
+                () -> {
+                    try {
+                        client.describeAcls(AclBindingFilter.ANY).values().get();
+                    } catch (ExecutionException e) {
+                        if (e.getCause() instanceof SecurityDisabledException)
+                            return;
+                        throw e.getCause();
                     }
                 });
         }
     }
 
+    private void createTopicsResultTest(AdminClient client, Collection<String> topics)
+            throws InterruptedException, ExecutionException {
+        while (true) {
+            try {
+                client.describeTopics(topics).all().get();
+                break;
+            } catch (ExecutionException e) {
+                if (e.getCause() instanceof UnknownTopicOrPartitionException)
+                    continue;
+                throw e;
+            }
+        }
+    }
+
+    private boolean topicExists(Collection<TopicListing> listings, String topicName) {
+        boolean foundTopic = false;
+        for (TopicListing listing : listings) {
+            if (listing.name().equals(topicName)) {
+                if (listing.isInternal())
+                    throw new KafkaException(String.format("Did not expect %s to be an internal topic.", topicName));
+                foundTopic = true;
+            }
+        }
+        return foundTopic;
+    }
+
     private static class OffsetsForTime {
         Map<TopicPartition, OffsetAndTimestamp> result;
 
@@ -384,18 +387,8 @@ public void testConsume(final long prodTimeMs) throws Throwable {
             }
             final OffsetsForTime offsetsForTime = new OffsetsForTime();
             tryFeature("offsetsForTimes", testConfig.offsetsForTimesSupported,
-                    new Invoker() {
-                        @Override
-                        public void invoke() {
-                            offsetsForTime.result = consumer.offsetsForTimes(timestampsToSearch);
-                        }
-                    },
-                    new ResultTester() {
-                        @Override
-                        public void test() {
-                            log.info("offsetsForTime = {}", offsetsForTime.result);
-                        }
-                    });
+                () -> offsetsForTime.result = consumer.offsetsForTimes(timestampsToSearch),
+                () -> log.info("offsetsForTime = {}", offsetsForTime.result));
             // Whether or not offsetsForTimes works, beginningOffsets and endOffsets
             // should work.
             consumer.beginningOffsets(timestampsToSearch.keySet());
@@ -486,11 +479,7 @@ private interface ResultTester {
     }
 
     private void tryFeature(String featureName, boolean supported, Invoker invoker) throws Throwable {
-        tryFeature(featureName, supported, invoker, new ResultTester() {
-                @Override
-                public void test() {
-                }
-            });
+        tryFeature(featureName, supported, invoker, () -> { });
     }
 
     private void tryFeature(String featureName, boolean supported, Invoker invoker, ResultTester resultTester)

tools/src/main/java/org/apache/kafka/tools/ToolsUtils.java

@@ -19,7 +19,6 @@
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 
-import java.util.Comparator;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -32,12 +31,7 @@ public class ToolsUtils {
     public static void printMetrics(Map<MetricName, ? extends Metric> metrics) {
         if (metrics != null && !metrics.isEmpty()) {
             int maxLengthOfDisplayName = 0;
-            TreeMap<String, Object> sortedMetrics = new TreeMap<>(new Comparator<String>() {
-                @Override
-                public int compare(String o1, String o2) {
-                    return o1.compareTo(o2);
-                }
-            });
+            TreeMap<String, Object> sortedMetrics = new TreeMap<>();
             for (Metric metric : metrics.values()) {
                 MetricName mName = metric.metricName();
                 String mergedName = mName.group() + ":" + mName.name() + ":" + mName.tags();

tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java

@@ -263,18 +263,15 @@ public static void main(String[] args) throws IOException {
         final AtomicBoolean isShuttingDown = new AtomicBoolean(false);
         final AtomicLong remainingMessages = new AtomicLong(maxMessages);
         final AtomicLong numMessagesProcessed = new AtomicLong(0);
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            @Override
-            public void run() {
-                isShuttingDown.set(true);
-                // Flush any remaining messages
-                producer.close();
-                synchronized (consumer) {
-                    consumer.close();
-                }
-                System.out.println(shutDownString(numMessagesProcessed.get(), remainingMessages.get(), transactionalId));
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            isShuttingDown.set(true);
+            // Flush any remaining messages
+            producer.close();
+            synchronized (consumer) {
+                consumer.close();
             }
-        });
+            System.out.println(shutDownString(numMessagesProcessed.get(), remainingMessages.get(), transactionalId));
+        }));
 
         try {
             Random random = new Random();

tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java

@@ -620,12 +620,7 @@ public static void main(String[] args) {
 
         try {
             final VerifiableConsumer consumer = createFromArgs(parser, args);
-            Runtime.getRuntime().addShutdownHook(new Thread() {
-                @Override
-                public void run() {
-                    consumer.close();
-                }
-            });
+            Runtime.getRuntime().addShutdownHook(new Thread(() -> consumer.close()));
 
             consumer.run();
         } catch (ArgumentParserException e) {

tools/src/main/java/org/apache/kafka/tools/VerifiableLog4jAppender.java

@@ -241,13 +241,10 @@ public static void main(String[] args) throws IOException {
         final VerifiableLog4jAppender appender = createFromArgs(args);
         boolean infinite = appender.maxMessages < 0;
 
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            @Override
-            public void run() {
-                // Trigger main thread to stop producing messages
-                appender.stopLogging = true;
-            }
-        });
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            // Trigger main thread to stop producing messages
+            appender.stopLogging = true;
+        }));
 
         long maxMessages = infinite ? Long.MAX_VALUE : appender.maxMessages;
         for (long i = 0; i < maxMessages; i++) {

tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java

@@ -517,22 +517,19 @@ public static void main(String[] args) {
             final long startMs = System.currentTimeMillis();
             ThroughputThrottler throttler = new ThroughputThrottler(producer.throughput, startMs);
 
-            Runtime.getRuntime().addShutdownHook(new Thread() {
-                @Override
-                public void run() {
-                    // Trigger main thread to stop producing messages
-                    producer.stopProducing = true;
+            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+                // Trigger main thread to stop producing messages
+                producer.stopProducing = true;
 
-                    // Flush any remaining messages
-                    producer.close();
+                // Flush any remaining messages
+                producer.close();
 
-                    // Print a summary
-                    long stopMs = System.currentTimeMillis();
-                    double avgThroughput = 1000 * ((producer.numAcked) / (double) (stopMs - startMs));
+                // Print a summary
+                long stopMs = System.currentTimeMillis();
+                double avgThroughput = 1000 * ((producer.numAcked) / (double) (stopMs - startMs));
 
-                    producer.printJson(new ToolData(producer.numSent, producer.numAcked, producer.throughput, avgThroughput));
-                }
-            });
+                producer.printJson(new ToolData(producer.numSent, producer.numAcked, producer.throughput, avgThroughput));
+            }));
 
             producer.run(throttler);
         } catch (ArgumentParserException e) {

tools/src/main/java/org/apache/kafka/trogdor/agent/Agent.java

@@ -147,18 +147,15 @@ public static void main(String[] args) throws Exception {
         log.info("Starting agent process.");
         final Agent agent = new Agent(platform, Scheduler.SYSTEM, restServer, resource);
         restServer.start(resource);
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            @Override
-            public void run() {
-                log.warn("Running agent shutdown hook.");
-                try {
-                    agent.beginShutdown();
-                    agent.waitForShutdown();
-                } catch (Exception e) {
-                    log.error("Got exception while running agent shutdown hook.", e);
-                }
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            log.warn("Running agent shutdown hook.");
+            try {
+                agent.beginShutdown();
+                agent.waitForShutdown();
+            } catch (Exception e) {
+                log.error("Got exception while running agent shutdown hook.", e);
             }
-        });
+        }));
         agent.waitForShutdown();
     }
 };

tools/src/main/java/org/apache/kafka/trogdor/agent/WorkerManager.java

@@ -317,21 +317,18 @@ public void createWorker(long workerId, String taskId, TaskSpec spec) throws Thr
                 return;
             }
             KafkaFutureImpl<String> haltFuture = new KafkaFutureImpl<>();
-            haltFuture.thenApply(new KafkaFuture.BaseFunction<String, Void>() {
-                @Override
-                public Void apply(String errorString) {
-                    if (errorString == null)
-                        errorString = "";
-                    if (errorString.isEmpty()) {
-                        log.info("{}: Worker {} is halting.", nodeName, worker);
-                    } else {
-                        log.info("{}: Worker {} is halting with error {}",
-                            nodeName, worker, errorString);
-                    }
-                    stateChangeExecutor.submit(
-                        new HandleWorkerHalting(worker, errorString, false));
-                    return null;
+            haltFuture.thenApply((KafkaFuture.BaseFunction<String, Void>) errorString -> {
+                if (errorString == null)
+                    errorString = "";
+                if (errorString.isEmpty()) {
+                    log.info("{}: Worker {} is halting.", nodeName, worker);
+                } else {
+                    log.info("{}: Worker {} is halting with error {}",
+                        nodeName, worker, errorString);
                 }
+                stateChangeExecutor.submit(
+                    new HandleWorkerHalting(worker, errorString, false));
+                return null;
             });
             try {
                 worker.taskWorker.start(platform, worker.status, haltFuture);

tools/src/main/java/org/apache/kafka/trogdor/coordinator/Coordinator.java

@@ -162,18 +162,15 @@ public static void main(String[] args) throws Exception {
         final Coordinator coordinator = new Coordinator(platform, Scheduler.SYSTEM,
             restServer, resource, ThreadLocalRandom.current().nextLong(0, Long.MAX_VALUE / 2));
         restServer.start(resource);
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            @Override
-            public void run() {
-                log.warn("Running coordinator shutdown hook.");
-                try {
-                    coordinator.beginShutdown(false);
-                    coordinator.waitForShutdown();
-                } catch (Exception e) {
-                    log.error("Got exception while running coordinator shutdown hook.", e);
-                }
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            log.warn("Running coordinator shutdown hook.");
+            try {
+                coordinator.beginShutdown(false);
+                coordinator.waitForShutdown();
+            } catch (Exception e) {
+                log.error("Got exception while running coordinator shutdown hook.", e);
             }
-        });
+        }));
         coordinator.waitForShutdown();
     }
 };

tools/src/main/java/org/apache/kafka/trogdor/rest/JsonRestServer.java

@@ -132,22 +132,19 @@ public int port() {
      */
     public void beginShutdown() {
         if (!shutdownExecutor.isShutdown()) {
-            shutdownExecutor.submit(new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                    try {
-                        log.info("Stopping REST server");
-                        jettyServer.stop();
-                        jettyServer.join();
-                        log.info("REST server stopped");
-                    } catch (Exception e) {
-                        log.error("Unable to stop REST server", e);
-                    } finally {
-                        jettyServer.destroy();
-                    }
-                    shutdownExecutor.shutdown();
-                    return null;
+            shutdownExecutor.submit((Callable<Void>) () -> {
+                try {
+                    log.info("Stopping REST server");
+                    jettyServer.stop();
+                    jettyServer.join();
+                    log.info("REST server stopped");
+                } catch (Exception e) {
+                    log.error("Unable to stop REST server", e);
+                } finally {
+                    jettyServer.destroy();
                 }
+                shutdownExecutor.shutdown();
+                return null;
             });
         }
     }