Disable loading lookups by default in CompactionTask

server/src/main/java/org/apache/druid/server/lookup/cache/LookupLoadingSpec.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableSet;
 import org.apache.druid.error.InvalidInput;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
@@ -94,19 +95,34 @@ public static LookupLoadingSpec createFromContext(Map<String, Object> context, L
       return defaultSpec;
     }
 
-    final Object lookupModeValue = context.get(LookupLoadingSpec.CTX_LOOKUP_LOADING_MODE);
+    final Object lookupModeValue = context.get(CTX_LOOKUP_LOADING_MODE);
     if (lookupModeValue == null) {
       return defaultSpec;
     }
 
-    final LookupLoadingSpec.Mode lookupLoadingMode = LookupLoadingSpec.Mode.valueOf(lookupModeValue.toString());
+    final LookupLoadingSpec.Mode lookupLoadingMode;
+    try {
+      lookupLoadingMode = LookupLoadingSpec.Mode.valueOf(lookupModeValue.toString());
+    }
+    catch (IllegalArgumentException e) {
+      throw InvalidInput.exception("Invalid value of %s[%s]. Allowed values are %s",
+                                   CTX_LOOKUP_LOADING_MODE, lookupModeValue.toString(), Arrays.asList(LookupLoadingSpec.Mode.values()));
+    }
+
+    if (lookupLoadingMode == Mode.NONE) {
+      return NONE;
+    } else if (lookupLoadingMode == Mode.ALL) {
+      return ALL;
+    } else if (lookupLoadingMode == Mode.ONLY_REQUIRED) {
+      Collection<String> lookupsToLoad;
+      try {
+        lookupsToLoad = (Collection<String>) context.get(CTX_LOOKUPS_TO_LOAD);
+      }
+      catch (ClassCastException e) {
+        throw InvalidInput.exception("Invalid value of %s[%s]. Please provide a comma-separated list of lookup names.",
+                                     CTX_LOOKUPS_TO_LOAD, context.get(CTX_LOOKUPS_TO_LOAD));
+      }
 
-    if (lookupLoadingMode == LookupLoadingSpec.Mode.NONE) {
-      return LookupLoadingSpec.NONE;
-    } else if (lookupLoadingMode == LookupLoadingSpec.Mode.ALL) {
-      return LookupLoadingSpec.ALL;
-    } else if (lookupLoadingMode == LookupLoadingSpec.Mode.ONLY_REQUIRED) {
-      Collection<String> lookupsToLoad = (Collection<String>) context.get(LookupLoadingSpec.CTX_LOOKUPS_TO_LOAD);
       if (lookupsToLoad == null || lookupsToLoad.isEmpty()) {
         throw InvalidInput.exception("Set of lookups to load cannot be %s for mode[ONLY_REQUIRED].", lookupsToLoad);
       }

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

@@ -21,13 +21,18 @@
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
 import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.StringUtils;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import java.util.Arrays;
 import java.util.Set;
 
+@RunWith(JUnitParamsRunner.class)
 public class LookupLoadingSpecTest
 {
   @Test
@@ -138,4 +143,43 @@ public void testCreateLookupLoadingSpecFromContext()
         )
     );
   }
+
+  @Test
+  @Parameters(
+      {
+          "NONE1",
+          "A",
+          "Random mode",
+          "all",
+          "only required",
+          "none"
+      }
+  )
+  public void testCreateLookupLoadingSpecFromInvalidModeInContext(String mode)
+  {
+    final DruidException exception = Assert.assertThrows(DruidException.class, () -> LookupLoadingSpec.createFromContext(
+        ImmutableMap.of(LookupLoadingSpec.CTX_LOOKUP_LOADING_MODE, mode), LookupLoadingSpec.ALL));
+    Assert.assertEquals(StringUtils.format("Invalid value of %s[%s]. Allowed values are [ALL, NONE, ONLY_REQUIRED]",
+                                           LookupLoadingSpec.CTX_LOOKUP_LOADING_MODE, mode), exception.getMessage());
+  }
+
+  @Test
+  @Parameters(
+      {
+          "foo bar",
+          "foo]"
+      }
+  )
+  public void testCreateLookupLoadingSpecFromInvalidLookupsInContext(Object lookupsToLoad)
+  {
+    final DruidException exception = Assert.assertThrows(DruidException.class, () ->
+        LookupLoadingSpec.createFromContext(
+            ImmutableMap.of(
+                LookupLoadingSpec.CTX_LOOKUPS_TO_LOAD, lookupsToLoad,
+                LookupLoadingSpec.CTX_LOOKUP_LOADING_MODE, LookupLoadingSpec.Mode.ONLY_REQUIRED),
+            LookupLoadingSpec.ALL)
+    );
+    Assert.assertEquals(StringUtils.format("Invalid value of %s[%s]. Please provide a comma-separated list of lookup names.",
+                                      LookupLoadingSpec.CTX_LOOKUPS_TO_LOAD, lookupsToLoad), exception.getMessage());
+  }
 }