diff --git a/docs/content/querying/lookups.md b/docs/content/querying/lookups.md index 23ef47f6549e..96ffd7f48cdf 100644 --- a/docs/content/querying/lookups.md +++ b/docs/content/querying/lookups.md @@ -323,6 +323,8 @@ It is possible to save the configuration across restarts such that a node will n |Property|Description|Default| |--------|-----------|-------| |`druid.lookup.snapshotWorkingDir`| Working path used to store snapshot of current lookup configuration, leaving this property null will disable snapshot/bootstrap utility|null| +|`druid.lookup.numLookupLoadingThreads`| Number of threads for loading the lookups in parallel on startup. This thread pool is destroyed once startup is done. It is not kept during the lifetime of the JVM|Available Processors / 2| +|`druid.lookup.enableLookupSyncOnStartup`|Enable the lookup synchronization process with coordinator on startup. The queryable nodes will fetch and load the lookups from the coordinator instead of waiting for the coordinator to load the lookups for them. Users may opt to disable this option if there are no lookups configured in the cluster.|true| ## Introspect a Lookup diff --git a/extensions-core/histogram/pom.xml b/extensions-core/histogram/pom.xml index 8eb5ae08db29..d6df9405bd5b 100644 --- a/extensions-core/histogram/pom.xml +++ b/extensions-core/histogram/pom.xml @@ -59,6 +59,13 @@ test-jar test + + io.druid + druid-server + ${project.parent.version} + test + test-jar + junit junit diff --git a/indexing-service/src/test/java/io/druid/indexing/common/TestUtils.java b/indexing-service/src/test/java/io/druid/indexing/common/TestUtils.java index 013e91b479f9..e34ea20fbc28 100644 --- a/indexing-service/src/test/java/io/druid/indexing/common/TestUtils.java +++ b/indexing-service/src/test/java/io/druid/indexing/common/TestUtils.java @@ -29,7 +29,7 @@ import io.druid.java.util.common.ISE; import io.druid.java.util.common.logger.Logger; import io.druid.math.expr.ExprMacroTable; -import io.druid.query.expression.TestExprMacroTable; +import io.druid.query.expression.TestExpressionMacroTable; import io.druid.segment.IndexIO; import io.druid.segment.IndexMergerV9; import io.druid.segment.column.ColumnConfig; @@ -74,7 +74,7 @@ public int columnCacheSizeBytes() jsonMapper.setInjectableValues( new InjectableValues.Std() - .addValue(ExprMacroTable.class.getName(), TestExprMacroTable.INSTANCE) + .addValue(ExprMacroTable.class.getName(), TestExpressionMacroTable.INSTANCE) .addValue(IndexIO.class, indexIO) .addValue(ObjectMapper.class, jsonMapper) .addValue(ChatHandlerProvider.class, new NoopChatHandlerProvider()) diff --git a/processing/src/main/java/io/druid/query/dimension/DimensionSpec.java b/processing/src/main/java/io/druid/query/dimension/DimensionSpec.java index 7a98ffd5edb8..33f4de0fdbf5 100644 --- a/processing/src/main/java/io/druid/query/dimension/DimensionSpec.java +++ b/processing/src/main/java/io/druid/query/dimension/DimensionSpec.java @@ -35,8 +35,7 @@ @JsonSubTypes.Type(name = "default", value = DefaultDimensionSpec.class), @JsonSubTypes.Type(name = "extraction", value = ExtractionDimensionSpec.class), @JsonSubTypes.Type(name = "regexFiltered", value = RegexFilteredDimensionSpec.class), - @JsonSubTypes.Type(name = "listFiltered", value = ListFilteredDimensionSpec.class), - @JsonSubTypes.Type(name = "lookup", value = LookupDimensionSpec.class) + @JsonSubTypes.Type(name = "listFiltered", value = ListFilteredDimensionSpec.class) }) public interface DimensionSpec extends Cacheable { diff --git a/processing/src/main/java/io/druid/query/extraction/ExtractionFn.java b/processing/src/main/java/io/druid/query/extraction/ExtractionFn.java index df6c86c4867f..5e6a3b95c0ac 100644 --- a/processing/src/main/java/io/druid/query/extraction/ExtractionFn.java +++ b/processing/src/main/java/io/druid/query/extraction/ExtractionFn.java @@ -24,7 +24,6 @@ import io.druid.guice.annotations.ExtensionPoint; import io.druid.java.util.common.Cacheable; import io.druid.query.lookup.LookupExtractionFn; -import io.druid.query.lookup.RegisteredLookupExtractionFn; import javax.annotation.Nullable; @@ -41,7 +40,6 @@ @JsonSubTypes.Type(name = "timeFormat", value = TimeFormatExtractionFn.class), @JsonSubTypes.Type(name = "identity", value = IdentityExtractionFn.class), @JsonSubTypes.Type(name = "lookup", value = LookupExtractionFn.class), - @JsonSubTypes.Type(name = "registeredLookup", value = RegisteredLookupExtractionFn.class), @JsonSubTypes.Type(name = "substring", value = SubstringDimExtractionFn.class), @JsonSubTypes.Type(name = "cascade", value = CascadeExtractionFn.class), @JsonSubTypes.Type(name = "stringFormat", value = StringFormatExtractionFn.class), diff --git a/processing/src/main/java/io/druid/query/lookup/LookupConfig.java b/processing/src/main/java/io/druid/query/lookup/LookupConfig.java index fea346b745fd..7f424c1e987c 100644 --- a/processing/src/main/java/io/druid/query/lookup/LookupConfig.java +++ b/processing/src/main/java/io/druid/query/lookup/LookupConfig.java @@ -26,11 +26,19 @@ public class LookupConfig { - @JsonProperty - private final String snapshotWorkingDir; + @JsonProperty("snapshotWorkingDir") + private String snapshotWorkingDir; + + @JsonProperty("enableLookupSyncOnStartup") + private boolean enableLookupSyncOnStartup = true; + + @JsonProperty("numLookupLoadingThreads") + private int numLookupLoadingThreads = Runtime.getRuntime().availableProcessors() / 2; /** - * @param snapshotWorkingDir working directory to store lookups snapshot file, passing null or empty string will disable the snapshot utility + * @param snapshotWorkingDir working directory to store lookups snapshot file, passing null or empty string will disable the snapshot utility + * @param numLookupLoadingThreads number of threads for loading the lookups as part of the synchronization process + * @param enableLookupSyncOnStartup decides whether the lookup synchronization process should be enabled at startup */ @JsonCreator public LookupConfig( @@ -45,6 +53,15 @@ public String getSnapshotWorkingDir() return snapshotWorkingDir; } + public int getNumLookupLoadingThreads() + { + return numLookupLoadingThreads; + } + + public boolean getEnableLookupSyncOnStartup() + { + return enableLookupSyncOnStartup; + } @Override public boolean equals(Object o) @@ -58,7 +75,9 @@ public boolean equals(Object o) LookupConfig that = (LookupConfig) o; - return getSnapshotWorkingDir().equals(that.getSnapshotWorkingDir()); + return snapshotWorkingDir.equals(that.snapshotWorkingDir) && + enableLookupSyncOnStartup == that.enableLookupSyncOnStartup && + numLookupLoadingThreads == that.numLookupLoadingThreads; } @@ -67,6 +86,8 @@ public String toString() { return "LookupConfig{" + "snapshotWorkingDir='" + getSnapshotWorkingDir() + '\'' + + " numLookupLoadingThreads='" + getNumLookupLoadingThreads() + '\'' + + " enableLookupSyncOnStartup='" + getEnableLookupSyncOnStartup() + '\'' + '}'; } } diff --git a/processing/src/test/java/io/druid/query/expression/TestExprMacroTable.java b/processing/src/test/java/io/druid/query/expression/TestExprMacroTable.java index b1319cbe0146..8527b65b839b 100644 --- a/processing/src/test/java/io/druid/query/expression/TestExprMacroTable.java +++ b/processing/src/test/java/io/druid/query/expression/TestExprMacroTable.java @@ -20,17 +20,7 @@ package io.druid.query.expression; import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableMap; import io.druid.math.expr.ExprMacroTable; -import io.druid.query.extraction.MapLookupExtractor; -import io.druid.query.lookup.LookupExtractor; -import io.druid.query.lookup.LookupExtractorFactory; -import io.druid.query.lookup.LookupExtractorFactoryContainer; -import io.druid.query.lookup.LookupIntrospectHandler; -import io.druid.query.lookup.LookupReferencesManager; -import org.easymock.EasyMock; - -import javax.annotation.Nullable; public class TestExprMacroTable extends ExprMacroTable { @@ -41,7 +31,6 @@ private TestExprMacroTable() super( ImmutableList.of( new LikeExprMacro(), - new LookupExprMacro(createTestLookupReferencesManager(ImmutableMap.of("foo", "xfoo"))), new RegexpExtractExprMacro(), new TimestampCeilExprMacro(), new TimestampExtractExprMacro(), @@ -55,52 +44,4 @@ private TestExprMacroTable() ) ); } - - /** - * Returns a mock {@link LookupReferencesManager} that has one lookup, "lookyloo". - */ - public static LookupReferencesManager createTestLookupReferencesManager(final ImmutableMap theLookup) - { - final LookupReferencesManager lookupReferencesManager = EasyMock.createMock(LookupReferencesManager.class); - EasyMock.expect(lookupReferencesManager.get(EasyMock.eq("lookyloo"))).andReturn( - new LookupExtractorFactoryContainer( - "v0", - new LookupExtractorFactory() - { - @Override - public boolean start() - { - throw new UnsupportedOperationException(); - } - - @Override - public boolean close() - { - throw new UnsupportedOperationException(); - } - - @Override - public boolean replaces(@Nullable final LookupExtractorFactory other) - { - throw new UnsupportedOperationException(); - } - - @Override - public LookupIntrospectHandler getIntrospectHandler() - { - throw new UnsupportedOperationException(); - } - - @Override - public LookupExtractor get() - { - return new MapLookupExtractor(theLookup, false); - } - } - ) - ).anyTimes(); - EasyMock.expect(lookupReferencesManager.get(EasyMock.not(EasyMock.eq("lookyloo")))).andReturn(null).anyTimes(); - EasyMock.replay(lookupReferencesManager); - return lookupReferencesManager; - } } diff --git a/processing/src/test/java/io/druid/query/lookup/LookupConfigTest.java b/processing/src/test/java/io/druid/query/lookup/LookupConfigTest.java index 836176e59135..c2e39bf0a0b3 100644 --- a/processing/src/test/java/io/druid/query/lookup/LookupConfigTest.java +++ b/processing/src/test/java/io/druid/query/lookup/LookupConfigTest.java @@ -32,13 +32,36 @@ public class LookupConfigTest { ObjectMapper mapper = TestHelper.getJsonMapper(); - @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test public void TestSerDesr() throws IOException { LookupConfig lookupConfig = new LookupConfig(temporaryFolder.newFile().getAbsolutePath()); - Assert.assertEquals(lookupConfig, mapper.reader(LookupConfig.class).readValue(mapper.writeValueAsString(lookupConfig))); + Assert.assertEquals( + lookupConfig, + mapper.reader(LookupConfig.class).readValue(mapper.writeValueAsString(lookupConfig)) + ); + } + + @Test + public void testSerdeWithNonDefaults() throws Exception + { + String json = "{\n" + + " \"enableLookupSyncOnStartup\": false,\n" + + " \"snapshotWorkingDir\": \"/tmp\",\n" + + " \"numLookupLoadingThreads\": 4 \n" + + "}\n"; + LookupConfig config = mapper.readValue( + mapper.writeValueAsString( + mapper.readValue(json, LookupConfig.class) + ), + LookupConfig.class + ); + + Assert.assertEquals("/tmp", config.getSnapshotWorkingDir()); + Assert.assertEquals(false, config.getEnableLookupSyncOnStartup()); + Assert.assertEquals(4, config.getNumLookupLoadingThreads()); } } diff --git a/processing/src/main/java/io/druid/query/dimension/LookupDimensionSpec.java b/server/src/main/java/io/druid/query/dimension/LookupDimensionSpec.java similarity index 100% rename from processing/src/main/java/io/druid/query/dimension/LookupDimensionSpec.java rename to server/src/main/java/io/druid/query/dimension/LookupDimensionSpec.java diff --git a/processing/src/main/java/io/druid/query/expression/LookupExprMacro.java b/server/src/main/java/io/druid/query/expression/LookupExprMacro.java similarity index 100% rename from processing/src/main/java/io/druid/query/expression/LookupExprMacro.java rename to server/src/main/java/io/druid/query/expression/LookupExprMacro.java diff --git a/server/src/main/java/io/druid/query/lookup/LookupModule.java b/server/src/main/java/io/druid/query/lookup/LookupModule.java index c8fdeaf68c0d..59c0d7f96c3d 100644 --- a/server/src/main/java/io/druid/query/lookup/LookupModule.java +++ b/server/src/main/java/io/druid/query/lookup/LookupModule.java @@ -25,6 +25,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.Module; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.jsontype.NamedType; import com.fasterxml.jackson.databind.module.SimpleModule; import com.google.common.base.Preconditions; import com.google.common.base.Strings; @@ -48,6 +49,7 @@ import io.druid.guice.annotations.Smile; import io.druid.initialization.DruidModule; import io.druid.java.util.common.logger.Logger; +import io.druid.query.dimension.LookupDimensionSpec; import io.druid.query.expression.LookupExprMacro; import io.druid.server.DruidNode; import io.druid.server.http.HostAndPortWithScheme; @@ -84,7 +86,11 @@ public static String getTierListenerPath(String tier) public List getJacksonModules() { return ImmutableList.of( - new SimpleModule("DruidLookupModule").registerSubtypes(MapLookupExtractorFactory.class) + new SimpleModule("DruidLookupModule").registerSubtypes(MapLookupExtractorFactory.class), + new SimpleModule().registerSubtypes( + new NamedType(LookupDimensionSpec.class, "lookup"), + new NamedType(RegisteredLookupExtractionFn.class, "registeredLookup") + ) ); } diff --git a/processing/src/main/java/io/druid/query/lookup/LookupReferencesManager.java b/server/src/main/java/io/druid/query/lookup/LookupReferencesManager.java similarity index 59% rename from processing/src/main/java/io/druid/query/lookup/LookupReferencesManager.java rename to server/src/main/java/io/druid/query/lookup/LookupReferencesManager.java index 1485fe725cae..961447ab356f 100644 --- a/processing/src/main/java/io/druid/query/lookup/LookupReferencesManager.java +++ b/server/src/main/java/io/druid/query/lookup/LookupReferencesManager.java @@ -20,6 +20,7 @@ package io.druid.query.lookup; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; @@ -28,21 +29,34 @@ import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; import com.metamx.emitter.EmittingLogger; +import com.metamx.http.client.response.FullResponseHolder; +import io.druid.client.coordinator.Coordinator; import io.druid.concurrent.Execs; import io.druid.concurrent.LifecycleLock; +import io.druid.discovery.DruidLeaderClient; import io.druid.guice.ManageLifecycle; import io.druid.guice.annotations.Json; import io.druid.java.util.common.ISE; +import io.druid.java.util.common.StringUtils; import io.druid.java.util.common.lifecycle.LifecycleStart; import io.druid.java.util.common.lifecycle.LifecycleStop; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.jboss.netty.handler.codec.http.HttpResponseStatus; import javax.annotation.Nullable; +import java.io.IOException; +import java.net.MalformedURLException; +import java.util.AbstractMap; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; @@ -79,21 +93,44 @@ public class LookupReferencesManager //for unit testing only private final boolean testMode; + private final DruidLeaderClient druidLeaderClient; + + private final ObjectMapper jsonMapper; + + private static final TypeReference> LOOKUPS_ALL_REFERENCE = + new TypeReference>() + { + }; + + private final LookupListeningAnnouncerConfig lookupListeningAnnouncerConfig; + + private final LookupConfig lookupConfig; + @Inject - public LookupReferencesManager(LookupConfig lookupConfig, @Json ObjectMapper objectMapper) + public LookupReferencesManager( + LookupConfig lookupConfig, @Json ObjectMapper objectMapper, + @Coordinator DruidLeaderClient druidLeaderClient, + LookupListeningAnnouncerConfig lookupListeningAnnouncerConfig + ) { - this(lookupConfig, objectMapper, false); + this(lookupConfig, objectMapper, druidLeaderClient, lookupListeningAnnouncerConfig, false); } @VisibleForTesting - LookupReferencesManager(LookupConfig lookupConfig, ObjectMapper objectMapper, boolean testMode) + LookupReferencesManager( + LookupConfig lookupConfig, ObjectMapper objectMapper, DruidLeaderClient druidLeaderClient, + LookupListeningAnnouncerConfig lookupListeningAnnouncerConfig, boolean testMode + ) { if (Strings.isNullOrEmpty(lookupConfig.getSnapshotWorkingDir())) { this.lookupSnapshotTaker = null; } else { this.lookupSnapshotTaker = new LookupSnapshotTaker(objectMapper, lookupConfig.getSnapshotWorkingDir()); } - + this.druidLeaderClient = druidLeaderClient; + this.jsonMapper = objectMapper; + this.lookupListeningAnnouncerConfig = lookupListeningAnnouncerConfig; + this.lookupConfig = lookupConfig; this.testMode = testMode; } @@ -103,43 +140,34 @@ public void start() if (!lifecycleLock.canStart()) { throw new ISE("can't start."); } - try { LOG.info("LookupReferencesManager is starting."); - - loadSnapshotAndInitStateRef(); - + loadAllLookupsAndInitStateRef(); if (!testMode) { mainThread = Execs.makeThread( "LookupReferencesManager-MainThread", - new Runnable() - { - @Override - public void run() - { - try { - - if (!lifecycleLock.awaitStarted()) { - LOG.error("WTF! lifecycle not started, lookup update notices will not be handled."); - return; - } + () -> { + try { + if (!lifecycleLock.awaitStarted()) { + LOG.error("WTF! lifecycle not started, lookup update notices will not be handled."); + return; + } - while (!Thread.interrupted() && lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) { - try { - handlePendingNotices(); - LockSupport.parkNanos(LookupReferencesManager.this, TimeUnit.MINUTES.toNanos(1)); - } - catch (Throwable t) { - LOG.makeAlert(t, "Error occured while lookup notice handling.").emit(); - } + while (!Thread.interrupted() && lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) { + try { + handlePendingNotices(); + LockSupport.parkNanos(LookupReferencesManager.this, TimeUnit.MINUTES.toNanos(1)); + } + catch (Throwable t) { + LOG.makeAlert(t, "Error occured while lookup notice handling.").emit(); } } - catch (Throwable t) { - LOG.error(t, "Error while waiting for lifecycle start. lookup updates notices will not be handled"); - } - finally { - LOG.info("Lookup Management loop exited, Lookup notices are not handled anymore."); - } + } + catch (Throwable t) { + LOG.error(t, "Error while waiting for lifecycle start. lookup updates notices will not be handled"); + } + finally { + LOG.info("Lookup Management loop exited, Lookup notices are not handled anymore."); } }, true @@ -279,7 +307,11 @@ public LookupsState getAllLookupsState() return new LookupsState<>(lookupUpdateState.lookupMap, lookupsToLoad, lookupsToDrop); } - private void updateToLoadAndDrop(List notices, Map lookupsToLoad, Set lookupsToDrop) + private void updateToLoadAndDrop( + List notices, + Map lookupsToLoad, + Set lookupsToDrop + ) { for (Notice notice : notices) { if (notice instanceof LoadNotice) { @@ -299,34 +331,168 @@ private void updateToLoadAndDrop(List notices, Map lookupMap) { if (lookupSnapshotTaker != null) { - List lookups = new ArrayList<>(lookupMap.size()); - for (Map.Entry e : lookupMap.entrySet()) { - lookups.add(new LookupBean(e.getKey(), null, e.getValue())); + lookupSnapshotTaker.takeSnapshot(getLookupBeanList(lookupMap)); + } + } + + private void loadAllLookupsAndInitStateRef() + { + List lookupBeanList = getLookupsListFromLookupConfig(); + if (lookupBeanList != null) { + startLookups(lookupBeanList); + } else { + LOG.info("No lookups to be loaded at this point"); + stateRef.set(new LookupUpdateState(ImmutableMap.of(), ImmutableList.of(), ImmutableList.of())); + } + } + + /** + * Returns a list of lookups from the coordinator if the coordinator is available. If it's not available, returns null. + * + * @param tier lookup tier name + * + * @return list of LookupBean objects, or null + */ + @Nullable + private List getLookupListFromCoordinator(String tier) + { + try { + final FullResponseHolder response = fetchLookupsForTier(tier); + List lookupBeanList = new ArrayList<>(); + if (!response.getStatus().equals(HttpResponseStatus.OK)) { + LOG.error( + "Error while fetching lookup code from Coordinator with status[%s] and content[%s]", + response.getStatus(), + response.getContent() + ); + return null; } - lookupSnapshotTaker.takeSnapshot(lookups); + // Older version of getSpecificTier returns a list of lookup names. + // Lookup loading is performed via snapshot if older version is present. + // This check is only for backward compatibility and should be removed in a future release + if (response.getContent().startsWith("[")) { + LOG.info("Failed to retrieve lookup information from coordinator. Attempting to load lookups using snapshot instead"); + return null; + } else { + Map lookupMap = jsonMapper.readValue( + response.getContent(), + LOOKUPS_ALL_REFERENCE + ); + lookupMap.forEach((k, v) -> lookupBeanList.add(new LookupBean(k, null, v))); + + } + return lookupBeanList; + } + catch (Exception e) { + LOG.error(e, "Error while trying to get lookup list from coordinator for tier[%s]", tier); + return null; } } - private void loadSnapshotAndInitStateRef() + /** + * Returns a list of lookups from the snapshot if the lookupsnapshottaker is configured. If it's not available, returns null. + * + * @return list of LookupBean objects, or null + */ + @Nullable + private List getLookupListFromSnapshot() { if (lookupSnapshotTaker != null) { - ImmutableMap.Builder builder = ImmutableMap.builder(); + return lookupSnapshotTaker.pullExistingSnapshot(); + } + return null; + } - final List lookupBeanList = lookupSnapshotTaker.pullExistingSnapshot(); - for (LookupBean lookupBean : lookupBeanList) { - LookupExtractorFactoryContainer container = lookupBean.getContainer(); + private List getLookupBeanList(Map lookupMap) + { + List lookups = new ArrayList<>(lookupMap.size()); + for (Map.Entry e : lookupMap.entrySet()) { + lookups.add(new LookupBean(e.getKey(), null, e.getValue())); + } + return lookups; + } - if (container.getLookupExtractorFactory().start()) { - builder.put(lookupBean.getName(), container); - } else { - throw new ISE("Failed to start lookup [%s]:[%s]", lookupBean.getName(), container); - } + private List getLookupsListFromLookupConfig() + { + List lookupBeanList; + if (lookupConfig.getEnableLookupSyncOnStartup()) { + String tier = lookupListeningAnnouncerConfig.getLookupTier(); + lookupBeanList = getLookupListFromCoordinator(tier); + if (lookupBeanList == null) { + LOG.info("Coordinator is unavailable. Loading saved snapshot instead"); + lookupBeanList = getLookupListFromSnapshot(); } + } else { + lookupBeanList = getLookupListFromSnapshot(); + } + return lookupBeanList; + } + private void startLookups(List lookupBeanList) + { + ImmutableMap.Builder builder = ImmutableMap.builder(); + ExecutorService executorService = Execs.multiThreaded( + lookupConfig.getNumLookupLoadingThreads(), + "LookupReferencesManager-Startup-%s" + ); + ExecutorCompletionService completionService = new ExecutorCompletionService(executorService); + List> futures = new ArrayList<>(); + try { + LOG.info("Starting lookup loading process"); + for (LookupBean lookupBean : lookupBeanList) { + futures.add( + completionService.submit( + () -> { + LookupExtractorFactoryContainer container = lookupBean.getContainer(); + LOG.info( + "Starting lookup [%s]:[%s]", + lookupBean.getName(), + container + ); + if (container.getLookupExtractorFactory().start()) { + LOG.info( + "Started lookup [%s]:[%s]", + lookupBean.getName(), + container + ); + return new AbstractMap.SimpleImmutableEntry<>(lookupBean.getName(), container); + } else { + LOG.error( + "Failed to start lookup [%s]:[%s]", + lookupBean.getName(), + container + ); + return null; + } + } + ) + ); + } + for (int i = 0; i < futures.size(); i++) { + try { + final Future> completedFuture = completionService + .take(); + final AbstractMap.SimpleImmutableEntry lookupResult = completedFuture + .get(); + if (lookupResult != null) { + builder.put(lookupResult); + } + } + catch (ExecutionException e) { + LOG.error(e, "Execution error during lookup loading."); + } + } stateRef.set(new LookupUpdateState(builder.build(), ImmutableList.of(), ImmutableList.of())); - } else { - stateRef.set(new LookupUpdateState(ImmutableMap.of(), ImmutableList.of(), ImmutableList.of())); + } + catch (Exception e) { + LOG.error(e, "Failed to finish lookup load process."); + for (Future future : futures) { + future.cancel(true); + } + } + finally { + executorService.shutdownNow(); } } @@ -341,6 +507,19 @@ private LookupUpdateState atomicallyUpdateStateRef(Function theLookup) + { + final LookupReferencesManager lookupReferencesManager = EasyMock.createMock(LookupReferencesManager.class); + EasyMock.expect(lookupReferencesManager.get(EasyMock.eq("lookyloo"))).andReturn( + new LookupExtractorFactoryContainer( + "v0", + new LookupExtractorFactory() + { + @Override + public boolean start() + { + throw new UnsupportedOperationException(); + } + + @Override + public boolean close() + { + throw new UnsupportedOperationException(); + } + + @Override + public boolean replaces(@Nullable final LookupExtractorFactory other) + { + throw new UnsupportedOperationException(); + } + + @Override + public LookupIntrospectHandler getIntrospectHandler() + { + throw new UnsupportedOperationException(); + } + + @Override + public LookupExtractor get() + { + return new MapLookupExtractor(theLookup, false); + } + } + ) + ).anyTimes(); + EasyMock.expect(lookupReferencesManager.get(EasyMock.not(EasyMock.eq("lookyloo")))).andReturn(null).anyTimes(); + EasyMock.replay(lookupReferencesManager); + return lookupReferencesManager; + } +} diff --git a/server/src/test/java/io/druid/query/lookup/LookupReferencesManagerTest.java b/server/src/test/java/io/druid/query/lookup/LookupReferencesManagerTest.java index 8aaf8b5c86ed..4395145b4edd 100644 --- a/server/src/test/java/io/druid/query/lookup/LookupReferencesManagerTest.java +++ b/server/src/test/java/io/druid/query/lookup/LookupReferencesManagerTest.java @@ -22,9 +22,15 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.ImmutableMap; import com.metamx.emitter.EmittingLogger; +import com.metamx.http.client.Request; +import com.metamx.http.client.response.FullResponseHolder; +import io.druid.discovery.DruidLeaderClient; import io.druid.jackson.DefaultObjectMapper; import io.druid.server.metrics.NoopServiceEmitter; import org.easymock.EasyMock; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.jboss.netty.handler.codec.http.HttpResponse; +import org.jboss.netty.handler.codec.http.HttpResponseStatus; import org.junit.Assert; import org.junit.Before; import org.junit.Rule; @@ -32,11 +38,35 @@ import org.junit.rules.TemporaryFolder; import java.io.IOException; +import java.net.URL; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.TimeUnit; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.reset; + public class LookupReferencesManagerTest { LookupReferencesManager lookupReferencesManager; + + private DruidLeaderClient druidLeaderClient; + + private LookupListeningAnnouncerConfig config; + + private static final String propertyBase = "some.property"; + + private static final String LOOKUP_TIER = "lookupTier"; + + private static final int LOOKUP_THREADS = 1; + + private static final boolean LOOKUP_DISABLE = false; + + LookupExtractorFactory lookupExtractorFactory; + + LookupExtractorFactoryContainer container; @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); ObjectMapper mapper = new DefaultObjectMapper(); @@ -46,22 +76,49 @@ public void setUp() throws IOException { EmittingLogger.registerEmitter(new NoopServiceEmitter()); + druidLeaderClient = EasyMock.createMock(DruidLeaderClient.class); + + config = createMock(LookupListeningAnnouncerConfig.class); + + lookupExtractorFactory = new MapLookupExtractorFactory( + ImmutableMap.of( + "key", + "value" + ), true + ); + container = new LookupExtractorFactoryContainer("v0", lookupExtractorFactory); mapper.registerSubtypes(MapLookupExtractorFactory.class); + String temporaryPath = temporaryFolder.newFolder().getAbsolutePath(); lookupReferencesManager = new LookupReferencesManager( new LookupConfig(temporaryFolder.newFolder().getAbsolutePath()), - mapper, + mapper, druidLeaderClient, config, true ); } @Test - public void testStartStop() + public void testStartStop() throws InterruptedException, IOException { lookupReferencesManager = new LookupReferencesManager( new LookupConfig(null), - mapper + mapper, druidLeaderClient, config ); + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForStartStop", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); Assert.assertFalse(lookupReferencesManager.lifecycleLock.awaitStarted(1, TimeUnit.MICROSECONDS)); Assert.assertNull(lookupReferencesManager.mainThread); Assert.assertNull(lookupReferencesManager.stateRef.get()); @@ -107,6 +164,22 @@ public void testAddGetRemove() throws Exception EasyMock.expect(lookupExtractorFactory.start()).andReturn(true).once(); EasyMock.expect(lookupExtractorFactory.close()).andReturn(true).once(); EasyMock.replay(lookupExtractorFactory); + + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForAddGetRemove", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); Assert.assertNull(lookupReferencesManager.get("test")); @@ -130,6 +203,21 @@ public void testCloseIsCalledAfterStopping() throws Exception EasyMock.expect(lookupExtractorFactory.start()).andReturn(true).once(); EasyMock.expect(lookupExtractorFactory.close()).andReturn(true).once(); EasyMock.replay(lookupExtractorFactory); + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForCloseIsCalledAfterStopping", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); lookupReferencesManager.add("testMock", new LookupExtractorFactoryContainer("0", lookupExtractorFactory)); lookupReferencesManager.handlePendingNotices(); @@ -146,6 +234,21 @@ public void testCloseIsCalledAfterRemove() throws Exception EasyMock.expect(lookupExtractorFactory.close()).andReturn(true).once(); EasyMock.replay(lookupExtractorFactory); + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForCloseIsCalledAfterRemove", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); lookupReferencesManager.add("testMock", new LookupExtractorFactoryContainer("0", lookupExtractorFactory)); lookupReferencesManager.handlePendingNotices(); @@ -157,8 +260,23 @@ public void testCloseIsCalledAfterRemove() throws Exception } @Test - public void testGetNotThere() + public void testGetNotThere() throws Exception { + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForGetNotThere", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); Assert.assertNull(lookupReferencesManager.get("notThere")); } @@ -174,7 +292,21 @@ public void testUpdateWithHigherVersion() throws Exception EasyMock.expect(lookupExtractorFactory2.start()).andReturn(true).once(); EasyMock.replay(lookupExtractorFactory1, lookupExtractorFactory2); - + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForUpdateWithHigherVersion", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); lookupReferencesManager.add("testName", new LookupExtractorFactoryContainer("1", lookupExtractorFactory1)); lookupReferencesManager.handlePendingNotices(); @@ -194,7 +326,21 @@ public void testUpdateWithLowerVersion() throws Exception LookupExtractorFactory lookupExtractorFactory2 = EasyMock.createNiceMock(LookupExtractorFactory.class); EasyMock.replay(lookupExtractorFactory1, lookupExtractorFactory2); - + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForUpdateWithLowerVersion", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); lookupReferencesManager.add("testName", new LookupExtractorFactoryContainer("1", lookupExtractorFactory1)); lookupReferencesManager.handlePendingNotices(); @@ -208,33 +354,24 @@ public void testUpdateWithLowerVersion() throws Exception @Test public void testRemoveNonExisting() throws Exception { - lookupReferencesManager.start(); - lookupReferencesManager.remove("test"); - lookupReferencesManager.handlePendingNotices(); - } - - @Test - public void testBootstrapFromFile() throws Exception - { - LookupExtractorFactory lookupExtractorFactory = new MapLookupExtractorFactory( - ImmutableMap.of( - "key", - "value" - ), true + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForRemoveNonExisting", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) ); - LookupExtractorFactoryContainer container = new LookupExtractorFactoryContainer("v0", lookupExtractorFactory); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); - lookupReferencesManager.add("testMockForBootstrap", container); + lookupReferencesManager.remove("test"); lookupReferencesManager.handlePendingNotices(); - lookupReferencesManager.stop(); - - lookupReferencesManager = new LookupReferencesManager( - new LookupConfig(lookupReferencesManager.lookupSnapshotTaker.getPersistFile().getParent()), - mapper, - true - ); - lookupReferencesManager.start(); - Assert.assertEquals(container, lookupReferencesManager.get("testMockForBootstrap")); } @Test @@ -269,7 +406,20 @@ public void testGetAllLookupsState() throws Exception ), true ) ); - + Map lookupMap = new HashMap<>(); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); lookupReferencesManager.add("one", container1); lookupReferencesManager.add("two", container2); @@ -290,14 +440,28 @@ public void testGetAllLookupsState() throws Exception Assert.assertTrue(state.getToDrop().contains("one")); } - @Test (timeout = 20000) + @Test(timeout = 20000) public void testRealModeWithMainThread() throws Exception { LookupReferencesManager lookupReferencesManager = new LookupReferencesManager( new LookupConfig(temporaryFolder.newFolder().getAbsolutePath()), - mapper + mapper, druidLeaderClient, config ); - + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForRealModeWithMainThread", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); lookupReferencesManager.start(); Assert.assertTrue(lookupReferencesManager.mainThread.isAlive()); @@ -324,4 +488,128 @@ public void testRealModeWithMainThread() throws Exception Assert.assertFalse(lookupReferencesManager.mainThread.isAlive()); } + + @Test + public void testCoordinatorLookupSync() throws Exception + { + LookupExtractorFactoryContainer container1 = new LookupExtractorFactoryContainer( + "0", + new MapLookupExtractorFactory( + ImmutableMap.of( + "key1", + "value1" + ), true + ) + ); + + LookupExtractorFactoryContainer container2 = new LookupExtractorFactoryContainer( + "0", + new MapLookupExtractorFactory( + ImmutableMap.of( + "key2", + "value2" + ), true + ) + ); + + LookupExtractorFactoryContainer container3 = new LookupExtractorFactoryContainer( + "0", + new MapLookupExtractorFactory( + ImmutableMap.of( + "key3", + "value3" + ), true + ) + ); + Map lookupMap = new HashMap<>(); + lookupMap.put("testLookup1", container1); + lookupMap.put("testLookup2", container2); + lookupMap.put("testLookup3", container3); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + replay(druidLeaderClient); + + lookupReferencesManager.start(); + Assert.assertEquals(container1, lookupReferencesManager.get("testLookup1")); + Assert.assertEquals(container2, lookupReferencesManager.get("testLookup2")); + Assert.assertEquals(container3, lookupReferencesManager.get("testLookup3")); + + } + + @Test + public void testLoadLookupOnCoordinatorFailure() throws Exception + { + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForLoadLookupOnCoordinatorFailure", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.NOT_FOUND, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andThrow(new IllegalStateException()); + replay(druidLeaderClient); + + lookupReferencesManager.start(); + lookupReferencesManager.add("testMockForLoadLookupOnCoordinatorFailure", container); + lookupReferencesManager.handlePendingNotices(); + lookupReferencesManager.stop(); + lookupReferencesManager = new LookupReferencesManager( + new LookupConfig(lookupReferencesManager.lookupSnapshotTaker.getPersistFile().getParent()), + mapper, druidLeaderClient, config, + true + ); + reset(config); + reset(druidLeaderClient); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + expect(druidLeaderClient.go(request)).andThrow(new IllegalStateException()); + replay(druidLeaderClient); + lookupReferencesManager.start(); + Assert.assertEquals(container, lookupReferencesManager.get("testMockForLoadLookupOnCoordinatorFailure")); + } + + @Test + public void testDisableLookupSync() throws Exception + { + LookupReferencesManager lookupReferencesManager = new LookupReferencesManager( + new LookupConfig(null), + mapper, druidLeaderClient, config + ); + Map lookupMap = new HashMap<>(); + lookupMap.put("testMockForDisableLookupSync", container); + String strResult = mapper.writeValueAsString(lookupMap); + Request request = new Request(HttpMethod.GET, new URL("http://localhost:1234/xx")); + expect(config.getLookupTier()).andReturn(LOOKUP_TIER); + replay(config); + expect(druidLeaderClient.makeRequest(HttpMethod.GET, "/druid/coordinator/v1/lookups/lookupTier?detailed=true")) + .andReturn(request); + FullResponseHolder responseHolder = new FullResponseHolder( + HttpResponseStatus.OK, + EasyMock.createNiceMock(HttpResponse.class), + new StringBuilder().append(strResult) + ); + expect(druidLeaderClient.go(request)).andReturn(responseHolder); + + lookupReferencesManager.start(); + Assert.assertNull(lookupReferencesManager.get("testMockForDisableLookupSync")); + } + } diff --git a/processing/src/test/java/io/druid/query/lookup/RegisteredLookupExtractionFnTest.java b/server/src/test/java/io/druid/query/lookup/RegisteredLookupExtractionFnTest.java similarity index 100% rename from processing/src/test/java/io/druid/query/lookup/RegisteredLookupExtractionFnTest.java rename to server/src/test/java/io/druid/query/lookup/RegisteredLookupExtractionFnTest.java diff --git a/server/src/test/java/io/druid/server/http/LookupCoordinatorResourceTest.java b/server/src/test/java/io/druid/server/http/LookupCoordinatorResourceTest.java index 499c236303b2..563542e86ad7 100644 --- a/server/src/test/java/io/druid/server/http/LookupCoordinatorResourceTest.java +++ b/server/src/test/java/io/druid/server/http/LookupCoordinatorResourceTest.java @@ -86,9 +86,10 @@ public InputStream openStream() throws IOException ImmutableMap.of(LOOKUP_NAME, SINGLE_LOOKUP), null, null ); - private static final Map> NODES_LOOKUP_STATE = ImmutableMap.of( - LOOKUP_NODE, LOOKUP_STATE - ); + private static final Map> NODES_LOOKUP_STATE = ImmutableMap + .of( + LOOKUP_NODE, LOOKUP_STATE + ); @Test public void testSimpleGet() @@ -208,6 +209,23 @@ public void testSimpleGetLookup() EasyMock.verify(lookupCoordinatorManager); } + @Test + public void testDetailedGetLookup() + { + final LookupCoordinatorManager lookupCoordinatorManager = EasyMock.createStrictMock(LookupCoordinatorManager.class); + EasyMock.expect(lookupCoordinatorManager.getKnownLookups()).andReturn(SINGLE_TIER_MAP).once(); + EasyMock.replay(lookupCoordinatorManager); + final LookupCoordinatorResource lookupCoordinatorResource = new LookupCoordinatorResource( + lookupCoordinatorManager, + mapper, + mapper + ); + final Response response = lookupCoordinatorResource.getSpecificTier(LOOKUP_TIER, true); + Assert.assertEquals(200, response.getStatus()); + Assert.assertEquals(SINGLE_TIER_MAP.get(LOOKUP_TIER), response.getEntity()); + EasyMock.verify(lookupCoordinatorManager); + } + @Test public void testMissingGetLookup() { @@ -765,7 +783,7 @@ public void testSimpleGetTier() mapper, mapper ); - final Response response = lookupCoordinatorResource.getSpecificTier(LOOKUP_TIER); + final Response response = lookupCoordinatorResource.getSpecificTier(LOOKUP_TIER, false); Assert.assertEquals(200, response.getStatus()); Assert.assertEquals(SINGLE_TIER_MAP.get(LOOKUP_TIER).keySet(), response.getEntity()); EasyMock.verify(lookupCoordinatorManager); @@ -785,7 +803,7 @@ public void testMissingGetTier() mapper, mapper ); - final Response response = lookupCoordinatorResource.getSpecificTier(tier); + final Response response = lookupCoordinatorResource.getSpecificTier(tier, false); Assert.assertEquals(404, response.getStatus()); EasyMock.verify(lookupCoordinatorManager); } @@ -801,7 +819,7 @@ public void testNullGetTier() mapper, mapper ); - final Response response = lookupCoordinatorResource.getSpecificTier(tier); + final Response response = lookupCoordinatorResource.getSpecificTier(tier, false); Assert.assertEquals(400, response.getStatus()); Assert.assertEquals(ImmutableMap.of("error", "`tier` required"), response.getEntity()); EasyMock.verify(lookupCoordinatorManager); @@ -819,7 +837,7 @@ public void testNullLookupsGetTier() mapper, mapper ); - final Response response = lookupCoordinatorResource.getSpecificTier(tier); + final Response response = lookupCoordinatorResource.getSpecificTier(tier, false); Assert.assertEquals(404, response.getStatus()); Assert.assertEquals(ImmutableMap.of("error", "No lookups found"), response.getEntity()); EasyMock.verify(lookupCoordinatorManager); @@ -838,7 +856,7 @@ public void testExceptionalGetTier() mapper, mapper ); - final Response response = lookupCoordinatorResource.getSpecificTier(tier); + final Response response = lookupCoordinatorResource.getSpecificTier(tier, false); Assert.assertEquals(500, response.getStatus()); Assert.assertEquals(ImmutableMap.of("error", errMsg), response.getEntity()); EasyMock.verify(lookupCoordinatorManager); diff --git a/sql/pom.xml b/sql/pom.xml index 5311c33c700a..1bbdf067d473 100644 --- a/sql/pom.xml +++ b/sql/pom.xml @@ -83,6 +83,13 @@ test-jar test + + io.druid + druid-server + ${project.parent.version} + test-jar + test + diff --git a/sql/src/test/java/io/druid/sql/calcite/util/CalciteTests.java b/sql/src/test/java/io/druid/sql/calcite/util/CalciteTests.java index e9d12f65ccad..27f4b69bc71f 100644 --- a/sql/src/test/java/io/druid/sql/calcite/util/CalciteTests.java +++ b/sql/src/test/java/io/druid/sql/calcite/util/CalciteTests.java @@ -57,7 +57,7 @@ import io.druid.query.aggregation.FloatSumAggregatorFactory; import io.druid.query.aggregation.hyperloglog.HyperUniquesAggregatorFactory; import io.druid.query.expression.LookupExprMacro; -import io.druid.query.expression.TestExprMacroTable; +import io.druid.query.expression.TestExpressionMacroTable; import io.druid.query.groupby.GroupByQuery; import io.druid.query.groupby.GroupByQueryConfig; import io.druid.query.groupby.GroupByQueryRunnerTest; @@ -140,7 +140,7 @@ public void configure(final Binder binder) binder.bind(LookupReferencesManager.class) .toInstance( - TestExprMacroTable.createTestLookupReferencesManager( + TestExpressionMacroTable.createTestLookupReferencesManager( ImmutableMap.of( "a", "xa", "abc", "xabc"