EmitterModule: Throw an error on invalid emitter types. (#7328)

* EmitterModule: Throw an error on invalid emitter types.

The current behavior of silently using the "noop" emitter is unhelpful
and makes it difficult to debug config typos.

* Add comments.

server/src/main/java/org/apache/druid/server/emitter/EmitterModule.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.server.emitter;
 
+import com.google.common.base.Strings;
 import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Binder;
@@ -130,13 +131,16 @@ public class EmitterModule implements Module
     {
       final List<Binding<Emitter>> emitterBindings = injector.findBindingsByType(new TypeLiteral<Emitter>(){});
 
-      emitter = findEmitter(emitterType, emitterBindings);
+      if (Strings.isNullOrEmpty(emitterType)) {
-
+        // If the emitter is unspecified, we want to default to the no-op emitter. Include empty string here too, just
-      if (emitter == null) {
+        // in case nulls are translated to empty strings at some point somewhere in the system.
         emitter = findEmitter(NoopEmitterModule.EMITTER_TYPE, emitterBindings);
+      } else {
+        emitter = findEmitter(emitterType, emitterBindings);
       }
 
       if (emitter == null) {
+        // If the requested emitter couldn't be found, throw an error. It might mean a typo, or a missing extension.
         List<String> knownTypes = new ArrayList<>();
         for (Binding<Emitter> binding : emitterBindings) {
           final Annotation annotation = binding.getKey().getAnnotation();

server/src/test/java/org/apache/druid/server/emitter/EmitterModuleTest.java

@@ -31,9 +31,13 @@ import org.apache.druid.guice.LifecycleModule;
 import org.apache.druid.guice.ServerModule;
 import org.apache.druid.jackson.JacksonModule;
 import org.apache.druid.java.util.emitter.core.Emitter;
+import org.apache.druid.java.util.emitter.core.NoopEmitter;
 import org.apache.druid.java.util.emitter.core.ParametrizedUriEmitter;
+import org.hamcrest.CoreMatchers;
 import org.junit.Assert;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import javax.validation.Validation;
 import javax.validation.Validator;
@@ -41,6 +45,8 @@ import java.util.Properties;
 
 public class EmitterModuleTest
 {
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
 
   @Test
   public void testParametrizedUriEmitterConfig()
@@ -55,10 +61,30 @@ public class EmitterModuleTest
     props.setProperty("druid.emitter.parametrized.httpEmitting.maxBatchSize", "4");
     props.setProperty("druid.emitter.parametrized.httpEmitting.flushTimeOut", "1000");
 
-    final Emitter emitter =
+    final Emitter emitter = makeInjectorWithProperties(props).getInstance(Emitter.class);
-        makeInjectorWithProperties(props).getInstance(Emitter.class);
+
     // Testing that ParametrizedUriEmitter is successfully deserialized from the above config
-    Assert.assertTrue(emitter instanceof ParametrizedUriEmitter);
+    Assert.assertThat(emitter, CoreMatchers.instanceOf(ParametrizedUriEmitter.class));
+  }
+
+  @Test
+  public void testMissingEmitterType()
+  {
+    final Properties props = new Properties();
+    props.setProperty("druid.emitter", "");
+
+    final Emitter emitter = makeInjectorWithProperties(props).getInstance(Emitter.class);
+    Assert.assertThat(emitter, CoreMatchers.instanceOf(NoopEmitter.class));
+  }
+
+  @Test
+  public void testInvalidEmitterType()
+  {
+    final Properties props = new Properties();
+    props.setProperty("druid.emitter", "invalid");
+
+    expectedException.expectMessage("Unknown emitter type[druid.emitter]=[invalid]");
+    makeInjectorWithProperties(props).getInstance(Emitter.class);
   }
 
   private Injector makeInjectorWithProperties(final Properties props)