[Refactor] Refactor some method to java8

pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java

@@ -76,7 +76,6 @@
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.Stat;
-import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -281,13 +280,10 @@ public void testUnloadNamespaceBundleFailure() throws Exception {
         CompletableFuture<Optional<Topic>> topicFuture = CompletableFuture.completedFuture(Optional.of(spyTopic));
         // add mock topic
         topics.put(topicName, topicFuture);
-        doAnswer(new Answer<CompletableFuture<Void>>() {
-            @Override
-            public CompletableFuture<Void> answer(InvocationOnMock invocation) {
-                CompletableFuture<Void> result = new CompletableFuture<>();
-                result.completeExceptionally(new RuntimeException("first time failed"));
-                return result;
-            }
+        doAnswer((Answer<CompletableFuture<Void>>) invocation -> {
+            CompletableFuture<Void> result = new CompletableFuture<>();
+            result.completeExceptionally(new RuntimeException("first time failed"));
+            return result;
         }).when(spyTopic).close(false);
         NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
 
@@ -319,12 +315,7 @@ public void testUnloadNamespaceBundleWithStuckTopic() throws Exception {
         // add mock topic
         topics.put(topicName, topicFuture);
         // return uncompleted future as close-topic result.
-        doAnswer(new Answer<CompletableFuture<Void>>() {
-            @Override
-            public CompletableFuture<Void> answer(InvocationOnMock invocation) throws Throwable {
-                return new CompletableFuture<Void>();
-            }
-        }).when(spyTopic).close(false);
+        doAnswer((Answer<CompletableFuture<Void>>) invocation -> new CompletableFuture<Void>()).when(spyTopic).close(false);
         NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
 
         // try to unload bundle whose topic will be stuck

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java

@@ -82,7 +82,6 @@
 import org.apache.pulsar.common.policies.data.LocalPolicies;
 import org.apache.pulsar.common.policies.data.SubscriptionStats;
 import org.apache.pulsar.common.policies.data.TopicStats;
-import org.apache.pulsar.common.util.SimpleTextOutputStream;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -1001,12 +1000,7 @@ public void testStuckTopicUnloading() throws Exception {
 
     @Test
     public void testMetricsProvider() throws IOException {
-        PrometheusRawMetricsProvider rawMetricsProvider = new PrometheusRawMetricsProvider() {
-            @Override
-            public void generate(SimpleTextOutputStream stream) {
-                stream.write("test_metrics{label1=\"xyz\"} 10 \n");
-            }
-        };
+        PrometheusRawMetricsProvider rawMetricsProvider = stream -> stream.write("test_metrics{label1=\"xyz\"} 10 \n");
         getPulsar().addPrometheusRawMetricsProvider(rawMetricsProvider);
         HttpClient httpClient = HttpClientBuilder.create().build();
         final String metricsEndPoint = getPulsar().getWebServiceAddress() + "/metrics";

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java

@@ -30,8 +30,6 @@
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.protocol.Commands;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
 import org.testng.annotations.Test;
 import static org.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
 import static org.mockito.ArgumentMatchers.any;
@@ -44,7 +42,6 @@
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
 
 @Slf4j
 @Test(groups = "broker")
@@ -163,14 +160,11 @@ public void testIsDuplicateWithFailure() {
 
         EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
 
-        doAnswer(new Answer() {
-            @Override
-            public Object answer(InvocationOnMock invocationOnMock) {
-                Object[] args = invocationOnMock.getArguments();
-                Runnable test = (Runnable) args[0];
-                test.run();
-                return null;
-            }
+        doAnswer(invocationOnMock -> {
+            Object[] args = invocationOnMock.getArguments();
+            Runnable test = (Runnable) args[0];
+            test.run();
+            return null;
         }).when(eventLoopGroup).submit(any(Runnable.class));
 
         BrokerService brokerService = mock(BrokerService.class);

pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java

@@ -906,14 +906,11 @@ public void testOrderingWithConsumerListener(boolean partitioned) throws Excepti
                 .topic(topic)
                 .subscriptionName(subName)
                 .subscriptionType(SubscriptionType.Key_Shared)
-                .messageListener(new MessageListener<Integer>() {
-                    @Override
-                    public void received(Consumer<Integer> consumer, Message<Integer> msg) {
-                        try {
-                            Thread.sleep(random.nextInt(5));
-                            received.add(msg);
-                        } catch (InterruptedException ignore) {
-                        }
+                .messageListener((MessageListener<Integer>) (consumer1, msg) -> {
+                    try {
+                        Thread.sleep(random.nextInt(5));
+                        received.add(msg);
+                    } catch (InterruptedException ignore) {
                     }
                 })
                 .subscribe();