Use StubServiceEmitter in tests (#15426)

* Use StubServiceEmitter in tests
* Remove unthrown exception from declaration

extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/KubernetesPeonClientTest.java

@@ -32,8 +32,7 @@ import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import org.apache.druid.indexing.common.task.NoopTask;
 import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.emitter.core.Event;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -41,8 +40,6 @@ import org.junit.jupiter.api.Test;
 import java.io.InputStream;
 import java.net.HttpURLConnection;
 import java.sql.Timestamp;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -59,22 +56,13 @@ public class KubernetesPeonClientTest
   private KubernetesMockServer server;
   private KubernetesClientApi clientApi;
   private KubernetesPeonClient instance;
-  private ServiceEmitter serviceEmitter;
+  private StubServiceEmitter serviceEmitter;
-  private Collection<Event> events;
 
   @BeforeEach
   public void setup()
   {
     clientApi = new TestKubernetesClient(this.client);
-    events = new ArrayList<>();
+    serviceEmitter = new StubServiceEmitter("service", "host");
-    serviceEmitter = new ServiceEmitter("service", "host", null)
-    {
-      @Override
-      public void emit(Event event)
-      {
-        events.add(event);
-      }
-    };
     instance = new KubernetesPeonClient(clientApi, NAMESPACE, false, serviceEmitter);
   }
 
@@ -102,7 +90,7 @@ public class KubernetesPeonClientTest
     Pod peonPod = instance.launchPeonJobAndWaitForStart(job, NoopTask.create(), 1, TimeUnit.SECONDS);
 
     Assertions.assertNotNull(peonPod);
-    Assertions.assertEquals(1, events.size());
+    Assertions.assertEquals(1, serviceEmitter.getEvents().size());
   }
 
   @Test

extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java

@@ -34,8 +34,6 @@ import org.apache.druid.indexing.common.config.TaskConfigBuilder;
 import org.apache.druid.indexing.common.task.IndexTask;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexTuningConfig;
-import org.apache.druid.java.util.emitter.core.Event;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.k8s.overlord.KubernetesTaskRunnerConfig;
 import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
 import org.apache.druid.k8s.overlord.common.JobResponse;
@@ -47,6 +45,7 @@ import org.apache.druid.k8s.overlord.common.PeonCommandContext;
 import org.apache.druid.k8s.overlord.common.PeonPhase;
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.log.StartupLoggingConfig;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -92,14 +91,7 @@ public class DruidPeonClientIntegrationTest
         new NamedType(IndexTask.IndexTuningConfig.class, "index")
     );
     k8sClient = new DruidKubernetesClient();
-    ServiceEmitter serviceEmitter = new ServiceEmitter("service", "host", null)
+    peonClient = new KubernetesPeonClient(k8sClient, "default", false, new NoopServiceEmitter());
-    {
-      @Override
-      public void emit(Event event)
-      {
-      }
-    };
-    peonClient = new KubernetesPeonClient(k8sClient, "default", false, serviceEmitter);
     druidNode = new DruidNode(
         "test",
         null,

extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java

@@ -52,8 +52,6 @@ import org.apache.druid.initialization.Initialization;
 import org.apache.druid.java.util.common.guava.Accumulators;
 import org.apache.druid.java.util.common.guava.Sequence;
 import org.apache.druid.java.util.common.guava.Sequences;
-import org.apache.druid.java.util.emitter.core.Event;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.query.BrokerParallelMergeConfig;
 import org.apache.druid.query.Query;
 import org.apache.druid.query.QueryPlus;
@@ -374,13 +372,7 @@ public class MovingAverageQueryTest extends InitializedNullHandlingTest
     );
 
     ClientQuerySegmentWalker walker = new ClientQuerySegmentWalker(
-        new ServiceEmitter("", "", null)
+        new NoopServiceEmitter(),
-        {
-          @Override
-          public void emit(Event event)
-          {
-          }
-        },
         baseClient,
         null /* local client; unused in this test, so pass in null */,
         warehouse,

processing/src/test/java/org/apache/druid/query/MetricsEmittingQueryProcessingPoolTest.java

@@ -20,17 +20,12 @@
 package org.apache.druid.query;
 
 import com.google.common.util.concurrent.ListeningExecutorService;
-import org.apache.druid.java.util.emitter.core.Emitter;
-import org.apache.druid.java.util.emitter.core.Event;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
-import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @SuppressWarnings("DoNotMock")
 public class MetricsEmittingQueryProcessingPoolTest
 {
@@ -41,24 +36,14 @@ public class MetricsEmittingQueryProcessingPoolTest
     Mockito.when(service.getQueueSize()).thenReturn(10);
     Mockito.when(service.getActiveTasks()).thenReturn(2);
     ExecutorServiceMonitor monitor = new ExecutorServiceMonitor();
-    List<Event> events = new ArrayList<>();
     MetricsEmittingQueryProcessingPool processingPool = new MetricsEmittingQueryProcessingPool(service, monitor);
     Assert.assertSame(service, processingPool.delegate());
 
-    ServiceEmitter serviceEmitter = new ServiceEmitter("service", "host", Mockito.mock(Emitter.class))
+    final StubServiceEmitter serviceEmitter = new StubServiceEmitter("service", "host");
-    {
-      @Override
-      public void emit(Event event)
-      {
-        events.add(event);
-      }
-    };
     monitor.doMonitor(serviceEmitter);
-    Assert.assertEquals(2, events.size());
+
-    Assert.assertEquals(((ServiceMetricEvent) (events.get(0))).getMetric(), "segment/scan/pending");
+    serviceEmitter.verifyValue("segment/scan/pending", 10);
-    Assert.assertEquals(((ServiceMetricEvent) (events.get(0))).getValue(), 10);
+    serviceEmitter.verifyValue("segment/scan/active", 2);
-    Assert.assertEquals(((ServiceMetricEvent) (events.get(1))).getMetric(), "segment/scan/active");
-    Assert.assertEquals(((ServiceMetricEvent) (events.get(1))).getValue(), 2);
   }
 
   @Test

server/src/test/java/org/apache/druid/client/cache/MemcachedCacheTest.java

@@ -54,11 +54,10 @@ import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.lifecycle.Lifecycle;
 import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.java.util.emitter.core.Emitter;
 import org.apache.druid.java.util.emitter.core.Event;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.metrics.AbstractMonitor;
-import org.easymock.EasyMock;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -67,7 +66,6 @@ import java.net.SocketAddress;
 import java.security.KeyManagementException;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -206,25 +204,16 @@ public class MemcachedCacheTest
   public void testMonitor() throws Exception
   {
     final MemcachedCache cache = MemcachedCache.create(memcachedCacheConfig);
-    final Emitter emitter = EasyMock.createNiceMock(Emitter.class);
+    final StubServiceEmitter serviceEmitter = new StubServiceEmitter("service", "host");
-    final Collection<Event> events = new ArrayList<>();
-    final ServiceEmitter serviceEmitter = new ServiceEmitter("service", "host", emitter)
-    {
-      @Override
-      public void emit(Event event)
-      {
-        events.add(event);
-      }
-    };
 
-    while (events.isEmpty()) {
+    while (serviceEmitter.getEvents().isEmpty()) {
       Thread.sleep(memcachedCacheConfig.getTimeout());
       cache.doMonitor(serviceEmitter);
     }
 
-    Assert.assertFalse(events.isEmpty());
+    Assert.assertFalse(serviceEmitter.getEvents().isEmpty());
     ObjectMapper mapper = new DefaultObjectMapper();
-    for (Event event : events) {
+    for (Event event : serviceEmitter.getEvents()) {
       log.debug("Found event `%s`", mapper.writeValueAsString(event.toMap()));
     }
   }

server/src/test/java/org/apache/druid/curator/DruidConnectionStateListenerTest.java

@@ -19,31 +19,22 @@
 
 package org.apache.druid.curator;
 
-import com.google.common.collect.ImmutableList;
-import com.google.errorprone.annotations.concurrent.GuardedBy;
 import org.apache.curator.framework.state.ConnectionState;
-import org.apache.druid.java.util.emitter.core.Event;
+import org.apache.druid.java.util.emitter.service.AlertEvent;
-import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
-import org.hamcrest.CoreMatchers;
-import org.hamcrest.MatcherAssert;
-import org.hamcrest.Matchers;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
 public class DruidConnectionStateListenerTest
 {
-  private TestEmitter emitter;
+  private StubServiceEmitter emitter;
   private DruidConnectionStateListener listener;
 
   @Before
   public void setUp()
   {
-    emitter = new TestEmitter();
+    emitter = new StubServiceEmitter("DruidConnectionStateListenerTest", "localhost");
     listener = new DruidConnectionStateListener(emitter);
   }
 
@@ -70,10 +61,7 @@ public class DruidConnectionStateListenerTest
   {
     listener.doMonitor(emitter);
     Assert.assertEquals(1, emitter.getEvents().size());
-
+    emitter.verifyValue("zk/connected", 0);
-    final Map<String, Object> eventMap = emitter.getEvents().get(0).toMap();
-    Assert.assertEquals("zk/connected", eventMap.get("metric"));
-    Assert.assertEquals(0, eventMap.get("value"));
   }
 
   @Test
@@ -83,9 +71,7 @@ public class DruidConnectionStateListenerTest
     listener.doMonitor(emitter);
     Assert.assertEquals(1, emitter.getEvents().size());
 
-    final Map<String, Object> eventMap = emitter.getEvents().get(0).toMap();
+    emitter.verifyValue("zk/connected", 1);
-    Assert.assertEquals("zk/connected", eventMap.get("metric"));
-    Assert.assertEquals(1, eventMap.get("value"));
   }
 
   @Test
@@ -95,9 +81,7 @@ public class DruidConnectionStateListenerTest
     listener.doMonitor(emitter);
     Assert.assertEquals(2, emitter.getEvents().size()); // 2 because stateChanged emitted an alert
 
-    final Map<String, Object> eventMap = emitter.getEvents().get(1).toMap();
+    emitter.verifyValue("zk/connected", 0);
-    Assert.assertEquals("zk/connected", eventMap.get("metric"));
-    Assert.assertEquals(0, eventMap.get("value"));
   }
 
   @Test
@@ -106,9 +90,9 @@ public class DruidConnectionStateListenerTest
     listener.stateChanged(null, ConnectionState.SUSPENDED);
     Assert.assertEquals(1, emitter.getEvents().size());
 
-    final Map<String, Object> alertMap = emitter.getEvents().get(0).toMap();
+    final AlertEvent alert = emitter.getAlerts().get(0);
-    Assert.assertEquals("alerts", alertMap.get("feed"));
+    Assert.assertEquals("alerts", alert.getFeed());
-    Assert.assertEquals("ZooKeeper connection[SUSPENDED]", alertMap.get("description"));
+    Assert.assertEquals("ZooKeeper connection[SUSPENDED]", alert.getDescription());
   }
 
   @Test
@@ -120,31 +104,8 @@ public class DruidConnectionStateListenerTest
     listener.stateChanged(null, ConnectionState.RECONNECTED);
     Assert.assertEquals(2, emitter.getEvents().size()); // the second stateChanged emits a metric
 
-    final Map<String, Object> eventMap = emitter.getEvents().get(1).toMap();
+    long observedReconnectTime = emitter.getValue("zk/reconnect/time", null).longValue();
-    Assert.assertEquals("metrics", eventMap.get("feed"));
+    Assert.assertTrue(observedReconnectTime >= 0);
-    Assert.assertEquals("zk/reconnect/time", eventMap.get("metric"));
-    MatcherAssert.assertThat(eventMap.get("value"), CoreMatchers.instanceOf(Long.class));
-    MatcherAssert.assertThat(((Number) eventMap.get("value")).longValue(), Matchers.greaterThanOrEqualTo(0L));
   }
 
-  private static class TestEmitter extends NoopServiceEmitter
-  {
-    @GuardedBy("events")
-    private final List<Event> events = new ArrayList<>();
-
-    @Override
-    public void emit(Event event)
-    {
-      synchronized (events) {
-        events.add(event);
-      }
-    }
-
-    public List<Event> getEvents()
-    {
-      synchronized (events) {
-        return ImmutableList.copyOf(events);
-      }
-    }
-  }
 }

server/src/test/java/org/apache/druid/server/initialization/jetty/JettyServerModuleTest.java

@@ -19,34 +19,16 @@
 
 package org.apache.druid.server.initialization.jetty;
 
-import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
-import org.apache.druid.java.util.emitter.core.Emitter;
-import org.apache.druid.java.util.emitter.core.Event;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
-import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
-import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
 public class JettyServerModuleTest
 {
   @Test
   public void testJettyServerModule()
   {
-    List<Event> events = new ArrayList<>();
-    ServiceEmitter serviceEmitter = new ServiceEmitter("service", "host", Mockito.mock(Emitter.class))
-    {
-      @Override
-      public void emit(Event event)
-      {
-        events.add(event);
-      }
-    };
     QueuedThreadPool jettyServerThreadPool = Mockito.mock(QueuedThreadPool.class);
     JettyServerModule.setJettyServerThreadPool(jettyServerThreadPool);
     Mockito.when(jettyServerThreadPool.getThreads()).thenReturn(100);
@@ -58,25 +40,17 @@ public class JettyServerModuleTest
     Mockito.when(jettyServerThreadPool.getBusyThreads()).thenReturn(60);
 
     JettyServerModule.JettyMonitor jettyMonitor = new JettyServerModule.JettyMonitor("ds", "t0");
+
+    final StubServiceEmitter serviceEmitter = new StubServiceEmitter("service", "host");
     jettyMonitor.doMonitor(serviceEmitter);
 
-    Assert.assertEquals(8, events.size());
+    serviceEmitter.verifyValue("jetty/numOpenConnections", 0);
-    List<Pair<String, Number>> expectedEvents = Arrays.asList(
+    serviceEmitter.verifyValue("jetty/threadPool/total", 100);
-        new Pair<>("jetty/numOpenConnections", 0),
+    serviceEmitter.verifyValue("jetty/threadPool/idle", 40);
-        new Pair<>("jetty/threadPool/total", 100),
+    serviceEmitter.verifyValue("jetty/threadPool/isLowOnThreads", 1);
-        new Pair<>("jetty/threadPool/idle", 40),
+    serviceEmitter.verifyValue("jetty/threadPool/min", 30);
-        new Pair<>("jetty/threadPool/isLowOnThreads", 1),
+    serviceEmitter.verifyValue("jetty/threadPool/max", 100);
-        new Pair<>("jetty/threadPool/min", 30),
+    serviceEmitter.verifyValue("jetty/threadPool/queueSize", 50);
-        new Pair<>("jetty/threadPool/max", 100),
+    serviceEmitter.verifyValue("jetty/threadPool/busy", 60);
-        new Pair<>("jetty/threadPool/queueSize", 50),
-        new Pair<>("jetty/threadPool/busy", 60)
-    );
-
-    for (int i = 0; i < expectedEvents.size(); i++) {
-      Pair<String, Number> expected = expectedEvents.get(i);
-      ServiceMetricEvent actual = (ServiceMetricEvent) (events.get(i));
-      Assert.assertEquals(expected.lhs, actual.getMetric());
-      Assert.assertEquals(expected.rhs, actual.getValue());
-    }
   }
 }

server/src/test/java/org/apache/druid/server/lookup/cache/LookupCoordinatorManagerTest.java

@@ -35,12 +35,10 @@ import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.emitter.core.Event;
-import org.apache.druid.java.util.emitter.core.LoggingEmitter;
-import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.http.client.HttpClient;
 import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
 import org.apache.druid.java.util.http.client.response.SequenceInputStreamResponseHandler;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
 import org.apache.druid.query.lookup.LookupsState;
 import org.apache.druid.server.http.HostAndPortWithScheme;
 import org.easymock.EasyMock;
@@ -48,9 +46,7 @@ import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.ExpectedException;
 
 import javax.ws.rs.core.Response;
 import java.io.ByteArrayInputStream;
@@ -61,13 +57,10 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class LookupCoordinatorManagerTest
 {
-  @Rule
-  public ExpectedException expectedException = ExpectedException.none();
   private final ObjectMapper mapper = new DefaultObjectMapper();
   private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider = EasyMock.createStrictMock(DruidNodeDiscoveryProvider.class);
   private final LookupNodeDiscovery lookupNodeDiscovery = EasyMock.createStrictMock(
@@ -111,31 +104,19 @@ public class LookupCoordinatorManagerTest
       Collections.emptySet()
   );
 
-  private static final AtomicLong EVENT_EMITS = new AtomicLong(0L);
+  private static StubServiceEmitter SERVICE_EMITTER;
-  private static ServiceEmitter SERVICE_EMITTER;
 
   @BeforeClass
   public static void setUpStatic()
   {
-    LoggingEmitter loggingEmitter = EasyMock.createNiceMock(LoggingEmitter.class);
+    SERVICE_EMITTER = new StubServiceEmitter("", "");
-    EasyMock.replay(loggingEmitter);
-    SERVICE_EMITTER = new ServiceEmitter("", "", loggingEmitter)
-    {
-      @Override
-      public void emit(Event event)
-      {
-        EVENT_EMITS.incrementAndGet();
-        super.emit(event);
-      }
-    };
     EmittingLogger.registerEmitter(SERVICE_EMITTER);
   }
 
   @Before
-  public void setUp() throws IOException
+  public void setUp()
   {
     SERVICE_EMITTER.flush();
-    EVENT_EMITS.set(0L);
 
     EasyMock.reset(lookupNodeDiscovery);
 
@@ -146,26 +127,22 @@ public class LookupCoordinatorManagerTest
             EasyMock.<TypeReference>anyObject(),
             EasyMock.<AtomicReference>isNull()
         )
-    ).andReturn(
+    ).andReturn(new AtomicReference<>(null)).anyTimes();
-        new AtomicReference<>(null)
-    ).anyTimes();
     EasyMock.expect(
         configManager.watch(
             EasyMock.eq(LookupCoordinatorManager.OLD_LOOKUP_CONFIG_KEY),
             EasyMock.<TypeReference>anyObject(),
             EasyMock.<AtomicReference>isNull()
         )
-    ).andReturn(
+    ).andReturn(new AtomicReference<>(null)).anyTimes();
-        new AtomicReference<>(null)
-    ).anyTimes();
     EasyMock.replay(configManager);
   }
 
   @After
-  public void tearDown() throws IOException
+  public void tearDown()
   {
+    Assert.assertEquals(0, SERVICE_EMITTER.getEvents().size());
     SERVICE_EMITTER.flush();
-    Assert.assertEquals(0, EVENT_EMITS.get());
   }
 
   @Test
@@ -546,8 +523,8 @@ public class LookupCoordinatorManagerTest
     };
     manager.start();
     final AuditInfo auditInfo = new AuditInfo("author", "comment", "localhost");
-    expectedException.expect(ISE.class);
+
-    manager.updateLookups(TIERED_LOOKUP_MAP_V0, auditInfo);
+    Assert.assertThrows(ISE.class, () -> manager.updateLookups(TIERED_LOOKUP_MAP_V0, auditInfo));
   }
 
   @Test