diff --git a/extensions-contrib/statsd-emitter/pom.xml b/extensions-contrib/statsd-emitter/pom.xml
index 326c45980fe..c83c261e039 100644
--- a/extensions-contrib/statsd-emitter/pom.xml
+++ b/extensions-contrib/statsd-emitter/pom.xml
@@ -41,7 +41,7 @@
     <dependency>
       <groupId>com.datadoghq</groupId>
       <artifactId>java-dogstatsd-client</artifactId>
-      <version>2.13.0</version>
+      <version>4.0.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.code.findbugs</groupId>
@@ -84,8 +84,8 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.easymock</groupId>
-      <artifactId>easymock</artifactId>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java
index b9392604637..84972f81e53 100644
--- a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java
+++ b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java
@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.timgroup.statsd.Event.AlertType;
 import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.NonBlockingStatsDClientBuilder;
 import com.timgroup.statsd.StatsDClient;
 import com.timgroup.statsd.StatsDClientErrorHandler;
 import org.apache.druid.java.util.common.StringUtils;
@@ -57,12 +58,12 @@ public class StatsDEmitter implements Emitter
 
   static StatsDEmitter of(StatsDEmitterConfig config, ObjectMapper mapper)
   {
-    NonBlockingStatsDClient client = new NonBlockingStatsDClient(
-        config.getPrefix(),
-        config.getHostname(),
-        config.getPort(),
-        config.isDogstatsd() ? config.getDogstatsdConstantTags().toArray(new String[0]) : EMPTY_ARRAY,
-        new StatsDClientErrorHandler()
+    NonBlockingStatsDClient client = new NonBlockingStatsDClientBuilder()
+        .prefix(config.getPrefix())
+        .hostname(config.getHostname())
+        .port(config.getPort())
+        .constantTags(config.isDogstatsd() ? config.getDogstatsdConstantTags().toArray(new String[0]) : EMPTY_ARRAY)
+        .errorHandler(new StatsDClientErrorHandler()
         {
           private int exceptionCount = 0;
 
@@ -74,8 +75,8 @@ public class StatsDEmitter implements Emitter
             }
             exceptionCount += 1;
           }
-        }
-    );
+        })
+        .build();
     return new StatsDEmitter(config, mapper, client);
   }
 
diff --git a/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java b/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java
index ebab54e47a5..0cd754a1b2b 100644
--- a/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java
+++ b/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java
@@ -20,6 +20,7 @@
 package org.apache.druid.emitter.statsd;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.timgroup.statsd.Event;
 import com.timgroup.statsd.StatsDClient;
@@ -27,62 +28,61 @@ import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.emitter.service.AlertBuilder;
 import org.apache.druid.java.util.emitter.service.AlertEvent;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
-import org.easymock.Capture;
-import org.easymock.EasyMock;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
 
 public class StatsDEmitterTest
 {
   @Test
   public void testConvertRange()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
     client.gauge("broker.query.cache.total.hitRate", 54);
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
                      .build(DateTimes.nowUtc(), "query/cache/total/hitRate", 0.54)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testConvertRangeWithDogstatsd()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, true, null, null, null),
         new ObjectMapper(),
         client
     );
     client.gauge("broker.query.cache.total.hitRate", 0.54);
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
                      .build(DateTimes.nowUtc(), "query/cache/total/hitRate", 0.54)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testNoConvertRange()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
     client.time("broker.query.time.data-source.groupBy", 10);
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
                      .setDimension("type", "groupBy")
@@ -97,20 +97,18 @@ public class StatsDEmitterTest
                      .build(DateTimes.nowUtc(), "query/time", 10)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testConfigOptions()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
     client.time("brokerHost1#broker#query#time#data-source#groupBy", 10);
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
                      .setDimension("type", "groupBy")
@@ -125,13 +123,12 @@ public class StatsDEmitterTest
                      .build(DateTimes.nowUtc(), "query/time", 10)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testDogstatsdEnabled()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, true, null, null, null),
         new ObjectMapper(),
@@ -140,7 +137,6 @@ public class StatsDEmitterTest
     client.time("broker#query#time", 10,
                 "dataSource:data-source", "type:groupBy", "hostname:brokerHost1"
     );
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
                      .setDimension("type", "groupBy")
@@ -155,32 +151,29 @@ public class StatsDEmitterTest
                      .build(DateTimes.nowUtc(), "query/time", 10)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testBlankHolderOptions()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
     client.count("brokerHost1.broker.jvm.gc.count.G1-GC", 1);
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("gcName", "G1 GC")
                      .build(DateTimes.nowUtc(), "jvm/gc/count", 1)
                      .build("broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testServiceAsTagOption()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
             new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, null),
             new ObjectMapper(),
@@ -189,20 +182,18 @@ public class StatsDEmitterTest
     client.time("druid.query.time", 10,
             "druid_service:druid/broker", "dataSource:data-source", "type:groupBy", "hostname:brokerHost1"
     );
-    EasyMock.replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
             .setDimension("dataSource", "data-source")
             .setDimension("type", "groupBy")
             .build(DateTimes.nowUtc(), "query/time", 10)
             .build("druid/broker", "brokerHost1")
     );
-    EasyMock.verify(client);
   }
 
   @Test
   public void testAlertEvent()
   {
-    StatsDClient client = EasyMock.createMock(StatsDClient.class);
+    StatsDClient client = mock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
         new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, true),
         new ObjectMapper(),
@@ -216,20 +207,20 @@ public class StatsDEmitterTest
         .withText("{\"exception\":\"NPE\"}")
         .build();
 
-    Capture<Event> eventCapture = EasyMock.newCapture();
-    client.recordEvent(
-        EasyMock.capture(eventCapture),
-        EasyMock.eq("feed:alerts"), EasyMock.eq("druid_service:druid/broker"),
-        EasyMock.eq("severity:anomaly"), EasyMock.eq("hostname:brokerHost1")
-    );
-    EasyMock.replay(client);
     emitter.emit(AlertBuilder.create("something bad happened [%s]", "exception")
                              .severity(AlertEvent.Severity.ANOMALY)
                              .addData(ImmutableMap.of("exception", "NPE"))
                              .build("druid/broker", "brokerHost1")
     );
-    EasyMock.verify(client);
-    Event actualEvent = eventCapture.getValue();
+
+    final ArgumentCaptor<Event> eventArgumentCaptor = ArgumentCaptor.forClass(Event.class);
+    verify(client).recordEvent(
+        eventArgumentCaptor.capture(),
+        eq("feed:alerts"), eq("druid_service:druid/broker"),
+        eq("severity:anomaly"), eq("hostname:brokerHost1")
+    );
+
+    Event actualEvent = eventArgumentCaptor.getValue();
     Assert.assertTrue(actualEvent.getMillisSinceEpoch() > 0);
     Assert.assertEquals(expectedEvent.getPriority(), actualEvent.getPriority());
     Assert.assertEquals(expectedEvent.getAlertType(), actualEvent.getAlertType());
@@ -237,6 +228,27 @@ public class StatsDEmitterTest
     Assert.assertEquals(expectedEvent.getText(), actualEvent.getText());
   }
 
+  @Test
+  public void testInitialization()
+  {
+    final StatsDEmitterConfig config = new StatsDEmitterConfig(
+        "localhost",
+        8888,
+        "druid",
+        "-",
+        true,
+        null,
+        null,
+        true,
+        ImmutableList.of("tag1", "value1"),
+        true,
+        true
+    );
+    try (StatsDEmitter emitter = StatsDEmitter.of(config, new ObjectMapper())) {
+
+    }
+  }
+
   @Test
   public void testJacksonModules()
   {