This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 093d60a0489 Remove usages of DruidLeaderClient from extensions (#18297)
093d60a0489 is described below
commit 093d60a04892b2181bc830caff7d3dca773c3357
Author: Kashif Faraz <[email protected]>
AuthorDate: Wed Jul 23 17:02:06 2025 +0530
Remove usages of DruidLeaderClient from extensions (#18297)
Description:
This is a follow up to the work done by @uds5501 in #18142 , #18192 and
#18195
to replace usages of `DruidLeaderClient` with a `ServiceClient`-based
implementation.
Changes:
- Remove usages of `DruidLeaderClient` in extensions `druid-basic-security`
and `druid-catalog`
- Use `ServiceClient` to make HTTP calls instead
- Remove classes `DruidLeaderClient`, `CoordinatorDiscoveryModule`
- Add embedded test `CatalogCompactionTest`
---
embedded-tests/pom.xml | 6 +
.../security/basic/BasicSecurityDruidModule.java | 15 +
.../security/basic/CoordinatorServiceClient.java | 52 ++++
...natorPollingBasicAuthenticatorCacheManager.java | 15 +-
...rdinatorPollingBasicAuthorizerCacheManager.java | 19 +-
...rPollingBasicAuthenticatorCacheManagerTest.java | 68 +++--
...atorPollingBasicAuthorizerCacheManagerTest.java | 116 ++++---
extensions-core/druid-catalog/pom.xml | 23 ++
.../apache/druid/catalog/sync/CatalogClient.java | 175 +++++++----
.../catalog/compact/CatalogCompactionTest.java | 163 ++++++++++
...posedAsBrokerQueryComponentSupplierWrapper.java | 2 -
.../apache/druid/discovery/DruidLeaderClient.java | 307 -------------------
.../druid/guice/CoordinatorDiscoveryModule.java | 58 ----
.../druid/initialization/CoreInjectorBuilder.java | 2 -
.../query/lookup/LookupReferencesManager.java | 24 +-
.../druid/server/http/ServletResourceUtils.java | 23 ++
.../druid/discovery/DruidLeaderClientTest.java | 338 ---------------------
.../server/http/ServletResourceUtilsTest.java | 40 +++
.../testing/embedded/ServerReferenceHolder.java | 10 +
.../testing/embedded/ServerReferencesProvider.java | 6 +
.../druid/testing/embedded/indexing/Resources.java | 0
.../druid/sql/calcite/schema/SystemSchema.java | 55 ----
.../schema/DruidCalciteSchemaModuleTest.java | 13 -
.../org/apache/druid/sql/guice/SqlModuleTest.java | 9 -
24 files changed, 578 insertions(+), 961 deletions(-)
diff --git a/embedded-tests/pom.xml b/embedded-tests/pom.xml
index 1392f15c806..fc36eee0eb5 100644
--- a/embedded-tests/pom.xml
+++ b/embedded-tests/pom.xml
@@ -153,6 +153,12 @@
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-catalog</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- Some embedded tests use the router and the web-console for debugging
-->
<dependency>
<groupId>org.apache.druid</groupId>
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicSecurityDruidModule.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicSecurityDruidModule.java
index fceaf7f2202..fa4d39d6c5c 100644
---
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicSecurityDruidModule.java
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicSecurityDruidModule.java
@@ -27,12 +27,16 @@ import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Provides;
import com.google.inject.name.Names;
+import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.guice.Jerseys;
import org.apache.druid.guice.JsonConfigProvider;
import org.apache.druid.guice.LazySingleton;
import org.apache.druid.guice.LifecycleModule;
+import org.apache.druid.guice.annotations.EscalatedGlobal;
import org.apache.druid.initialization.DruidModule;
import org.apache.druid.java.util.http.client.NettyHttpClient;
+import org.apache.druid.rpc.ServiceClientFactory;
+import org.apache.druid.rpc.ServiceLocator;
import org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator;
import org.apache.druid.security.basic.authentication.BasicHTTPEscalator;
import
org.apache.druid.security.basic.authentication.db.cache.BasicAuthenticatorCacheManager;
@@ -207,6 +211,17 @@ public class BasicSecurityDruidModule implements
DruidModule
NoopBasicAuthorizerCacheNotifier.class
);
}
+
+ @Provides
+ @LazySingleton
+ public static CoordinatorServiceClient createCoordinatorServiceClient(
+ @EscalatedGlobal ServiceClientFactory clientFactory,
+ @Coordinator ServiceLocator serviceLocator,
+ BasicAuthCommonCacheConfig cacheConfig
+ )
+ {
+ return new CoordinatorServiceClient(clientFactory, serviceLocator,
cacheConfig);
+ }
@Override
public List<? extends Module> getJacksonModules()
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/CoordinatorServiceClient.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/CoordinatorServiceClient.java
new file mode 100644
index 00000000000..171bfe5d2c5
--- /dev/null
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/CoordinatorServiceClient.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.security.basic;
+
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.rpc.ServiceClient;
+import org.apache.druid.rpc.ServiceClientFactory;
+import org.apache.druid.rpc.ServiceLocator;
+import org.apache.druid.rpc.StandardRetryPolicy;
+
+/**
+ * Wrapper over {@link ServiceClient} used to query the Coordinator.
+ */
+public class CoordinatorServiceClient
+{
+ private final ServiceClient serviceClient;
+
+ public CoordinatorServiceClient(
+ ServiceClientFactory clientFactory,
+ ServiceLocator coordinatorServiceLocator,
+ BasicAuthCommonCacheConfig cacheConfig
+ )
+ {
+ this.serviceClient = clientFactory.makeClient(
+ NodeRole.COORDINATOR.getJsonName(),
+ coordinatorServiceLocator,
+
StandardRetryPolicy.builder().maxAttempts(cacheConfig.getMaxSyncRetries()).build()
+ );
+ }
+
+ public ServiceClient getServiceClient()
+ {
+ return serviceClient;
+ }
+}
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
index 51758b9ec6e..3be5ca5b936 100644
---
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
@@ -24,9 +24,7 @@ import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Injector;
import org.apache.commons.lang3.ArrayUtils;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.concurrent.LifecycleLock;
-import org.apache.druid.discovery.DruidLeaderClient;
import org.apache.druid.guice.ManageLifecycle;
import org.apache.druid.guice.annotations.Smile;
import org.apache.druid.java.util.common.FileUtils;
@@ -38,11 +36,12 @@ import
org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.http.client.Request;
import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
+import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
import org.apache.druid.security.basic.BasicAuthUtils;
+import org.apache.druid.security.basic.CoordinatorServiceClient;
import org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator;
import
org.apache.druid.security.basic.authentication.entity.BasicAuthenticatorUser;
import org.apache.druid.server.security.Authenticator;
@@ -74,7 +73,7 @@ public class CoordinatorPollingBasicAuthenticatorCacheManager
implements BasicAu
private final Injector injector;
private final ObjectMapper objectMapper;
private final LifecycleLock lifecycleLock = new LifecycleLock();
- private final DruidLeaderClient druidLeaderClient;
+ private final CoordinatorServiceClient coordinatorClient;
private final BasicAuthCommonCacheConfig commonCacheConfig;
private final ScheduledExecutorService exec;
@@ -83,7 +82,7 @@ public class CoordinatorPollingBasicAuthenticatorCacheManager
implements BasicAu
Injector injector,
BasicAuthCommonCacheConfig commonCacheConfig,
@Smile ObjectMapper objectMapper,
- @Coordinator DruidLeaderClient druidLeaderClient
+ CoordinatorServiceClient coordinatorClient
)
{
this.exec =
Execs.scheduledSingleThreaded("BasicAuthenticatorCacheManager-Exec--%d");
@@ -92,7 +91,7 @@ public class CoordinatorPollingBasicAuthenticatorCacheManager
implements BasicAu
this.objectMapper = objectMapper;
this.cachedUserMaps = new ConcurrentHashMap<>();
this.authenticatorPrefixes = new HashSet<>();
- this.druidLeaderClient = druidLeaderClient;
+ this.coordinatorClient = coordinatorClient;
}
@LifecycleStart
@@ -254,11 +253,11 @@ public class
CoordinatorPollingBasicAuthenticatorCacheManager implements BasicAu
private Map<String, BasicAuthenticatorUser>
tryFetchUserMapFromCoordinator(String prefix) throws Exception
{
Map<String, BasicAuthenticatorUser> userMap = null;
- Request req = druidLeaderClient.makeRequest(
+ RequestBuilder req = new RequestBuilder(
HttpMethod.GET,
StringUtils.format("/druid-ext/basic-security/authentication/db/%s/cachedSerializedUserMap",
prefix)
);
- BytesFullResponseHolder responseHolder = druidLeaderClient.go(
+ BytesFullResponseHolder responseHolder =
coordinatorClient.getServiceClient().request(
req,
new BytesFullResponseHandler()
);
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
index 4bebc38b210..ee53f6e9c3e 100644
---
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
@@ -23,9 +23,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Injector;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.concurrent.LifecycleLock;
-import org.apache.druid.discovery.DruidLeaderClient;
import org.apache.druid.guice.ManageLifecycle;
import org.apache.druid.guice.annotations.Smile;
import org.apache.druid.java.util.common.FileUtils;
@@ -37,11 +35,12 @@ import
org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.http.client.Request;
import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
+import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
import org.apache.druid.security.basic.BasicAuthUtils;
+import org.apache.druid.security.basic.CoordinatorServiceClient;
import org.apache.druid.security.basic.authorization.BasicRoleBasedAuthorizer;
import
org.apache.druid.security.basic.authorization.entity.BasicAuthorizerGroupMapping;
import
org.apache.druid.security.basic.authorization.entity.BasicAuthorizerRole;
@@ -79,7 +78,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
private final Injector injector;
private final ObjectMapper objectMapper;
private final LifecycleLock lifecycleLock = new LifecycleLock();
- private final DruidLeaderClient druidLeaderClient;
+ private final CoordinatorServiceClient coordinatorClient;
private final BasicAuthCommonCacheConfig commonCacheConfig;
private final ScheduledExecutorService exec;
@@ -88,7 +87,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
Injector injector,
BasicAuthCommonCacheConfig commonCacheConfig,
@Smile ObjectMapper objectMapper,
- @Coordinator DruidLeaderClient druidLeaderClient
+ CoordinatorServiceClient coordinatorClient
)
{
this.exec =
Execs.scheduledSingleThreaded("CoordinatorPollingBasicAuthorizerCacheManager-Exec--%d");
@@ -100,7 +99,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
this.cachedGroupMappingMaps = new ConcurrentHashMap<>();
this.cachedGroupMappingRoleMaps = new ConcurrentHashMap<>();
this.authorizerPrefixes = new HashSet<>();
- this.druidLeaderClient = druidLeaderClient;
+ this.coordinatorClient = coordinatorClient;
}
@LifecycleStart
@@ -397,11 +396,11 @@ public class
CoordinatorPollingBasicAuthorizerCacheManager implements BasicAutho
String prefix
) throws Exception
{
- Request req = druidLeaderClient.makeRequest(
+ RequestBuilder req = new RequestBuilder(
HttpMethod.GET,
StringUtils.format("/druid-ext/basic-security/authorization/db/%s/cachedSerializedUserMap",
prefix)
);
- BytesFullResponseHolder responseHolder = druidLeaderClient.go(
+ BytesFullResponseHolder responseHolder =
coordinatorClient.getServiceClient().request(
req,
new BytesFullResponseHandler()
);
@@ -421,11 +420,11 @@ public class
CoordinatorPollingBasicAuthorizerCacheManager implements BasicAutho
String prefix
) throws Exception
{
- Request req = druidLeaderClient.makeRequest(
+ RequestBuilder req = new RequestBuilder(
HttpMethod.GET,
StringUtils.format("/druid-ext/basic-security/authorization/db/%s/cachedSerializedGroupMappingMap",
prefix)
);
- BytesFullResponseHolder responseHolder = druidLeaderClient.go(
+ BytesFullResponseHolder responseHolder =
coordinatorClient.getServiceClient().request(
req,
new BytesFullResponseHandler()
);
diff --git
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
index f06d2deeb04..af041ad5a35 100644
---
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
+++
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
@@ -20,22 +20,29 @@
package org.apache.druid.security.basic.authentication.db.cache;
import com.google.inject.Injector;
-import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.http.client.Request;
-import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
-import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.rpc.MockServiceClient;
+import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
+import org.apache.druid.security.basic.CoordinatorServiceClient;
import org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.security.AuthenticatorMapper;
import org.easymock.EasyMock;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.jboss.netty.handler.codec.http.HttpVersion;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -56,35 +63,42 @@ public class
CoordinatorPollingBasicAuthenticatorCacheManagerTest
.andReturn(new AuthenticatorMapper(Map.of("test-basic-auth",
authenticator))).once();
// Create a mock leader client and request
- final DruidLeaderClient leaderClient =
EasyMock.createStrictMock(DruidLeaderClient.class);
- final Request request = EasyMock.createStrictMock(Request.class);
+ final CoordinatorServiceClient leaderClient =
EasyMock.createStrictMock(CoordinatorServiceClient.class);
+ final MockServiceClient serviceClient = new MockServiceClient();
+
EasyMock.expect(leaderClient.getServiceClient()).andReturn(serviceClient).anyTimes();
// Return the first request immediately
- EasyMock.expect(leaderClient.makeRequest(EasyMock.anyObject(),
EasyMock.anyString()))
- .andReturn(request).once();
- EasyMock.expect(
- leaderClient.go(EasyMock.anyObject(),
EasyMock.anyObject(BytesFullResponseHandler.class))
- ).andReturn(
- new BytesFullResponseHolder(null)
- ).once();
+ final String path = StringUtils.format(
+
"/druid-ext/basic-security/authentication/db/%s/cachedSerializedUserMap",
+ "test-basic-auth"
+ );
+ serviceClient.expectAndRespond(
+ new RequestBuilder(HttpMethod.GET, path),
+ HttpResponseStatus.OK,
+ Map.of(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON),
+ new byte[0]
+ );
// Block the second request so that it can be interrupted by stop()
final AtomicBoolean isInterrupted = new AtomicBoolean(false);
- EasyMock.expect(leaderClient.makeRequest(EasyMock.anyObject(),
EasyMock.anyString()))
- .andReturn(request).once();
- EasyMock.expect(
- leaderClient.go(EasyMock.anyObject(),
EasyMock.anyObject(BytesFullResponseHandler.class))
- ).andAnswer(() -> {
- try {
- Thread.sleep(10_000);
- return null;
- }
- catch (InterruptedException e) {
- isInterrupted.set(true);
- throw e;
- }
- }).once();
+ serviceClient.expectAndRespond(
+ new RequestBuilder(HttpMethod.GET, path),
+ new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK) {
+ @Override
+ public ChannelBuffer getContent()
+ {
+ try {
+ Thread.sleep(10_000);
+ return null;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ );
EasyMock.replay(injector, leaderClient);
diff --git
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
index 602f90e4461..115132330e2 100644
---
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
+++
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
@@ -21,14 +21,15 @@ package
org.apache.druid.security.basic.authorization.db.cache;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Injector;
-import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.jackson.JacksonUtils;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.java.util.http.client.Request;
-import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.rpc.MockServiceClient;
+import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
+import org.apache.druid.security.basic.CoordinatorServiceClient;
import org.apache.druid.security.basic.authorization.BasicRoleBasedAuthorizer;
import
org.apache.druid.security.basic.authorization.entity.GroupMappingAndRoleMap;
import org.apache.druid.security.basic.authorization.entity.UserAndRoleMap;
@@ -36,7 +37,10 @@ import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.security.AuthorizerMapper;
import org.easymock.EasyMock;
import org.easymock.IAnswer;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+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.jboss.netty.handler.codec.http.HttpVersion;
@@ -53,14 +57,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class CoordinatorPollingBasicAuthorizerCacheManagerTest
{
private static final ObjectMapper MAPPER = TestHelper.JSON_MAPPER;
+ private static final String AUTHORIZER_NAME = "test-basic-auth";
@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder();
// Mocks
- private Request request;
private Injector injector;
- private DruidLeaderClient leaderClient;
+ private CoordinatorServiceClient leaderClient;
+ private MockServiceClient serviceClient;
private CoordinatorPollingBasicAuthorizerCacheManager manager;
@@ -72,12 +77,13 @@ public class
CoordinatorPollingBasicAuthorizerCacheManagerTest
final BasicRoleBasedAuthorizer authorizer =
EasyMock.createStrictMock(BasicRoleBasedAuthorizer.class);
injector = EasyMock.createStrictMock(Injector.class);
EasyMock.expect(injector.getInstance(AuthorizerMapper.class))
- .andReturn(new AuthorizerMapper(Map.of("test-basic-auth",
authorizer))).once();
+ .andReturn(new AuthorizerMapper(Map.of(AUTHORIZER_NAME,
authorizer))).once();
- request = EasyMock.createStrictMock(Request.class);
- leaderClient = EasyMock.createStrictMock(DruidLeaderClient.class);
+ serviceClient = new MockServiceClient();
+ leaderClient = EasyMock.createStrictMock(CoordinatorServiceClient.class);
+
EasyMock.expect(leaderClient.getServiceClient()).andReturn(serviceClient).anyTimes();
- final int numRetries = 10;
+ final int numRetries = 1;
manager = new CoordinatorPollingBasicAuthorizerCacheManager(
injector,
new BasicAuthCommonCacheConfig(0L, 1L,
temporaryFolder.newFolder().getAbsolutePath(), numRetries),
@@ -107,21 +113,24 @@ public class
CoordinatorPollingBasicAuthorizerCacheManagerTest
groupResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
GroupMappingAndRoleMap(Map.of(), Map.of())));
// Return the first set of requests immediately
- expectHttpRequestAndAnswer(() -> userResponseHolder);
- expectHttpRequestAndAnswer(() -> groupResponseHolder);
+ expectHttpRequestAndAnswer("cachedSerializedUserMap", () ->
userResponseHolder);
+ expectHttpRequestAndAnswer("cachedSerializedGroupMappingMap", () ->
groupResponseHolder);
// Block the second user request so that it can be interrupted by stop()
final AtomicBoolean isInterrupted = new AtomicBoolean(false);
- expectHttpRequestAndAnswer(() -> {
- try {
- Thread.sleep(10_000);
- return userResponseHolder;
- }
- catch (InterruptedException e) {
- isInterrupted.set(true);
- throw e;
- }
- });
+ expectHttpRequestAndAnswer(
+ "cachedSerializedUserMap",
+ () -> {
+ try {
+ Thread.sleep(10_000);
+ return userResponseHolder;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw e;
+ }
+ }
+ );
replayAll();
@@ -149,24 +158,27 @@ public class
CoordinatorPollingBasicAuthorizerCacheManagerTest
groupResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
GroupMappingAndRoleMap(Map.of(), Map.of())));
// Return the first set of requests immediately
- expectHttpRequestAndAnswer(() -> userResponseHolder);
- expectHttpRequestAndAnswer(() -> groupResponseHolder);
+ expectHttpRequestAndAnswer("cachedSerializedUserMap", () ->
userResponseHolder);
+ expectHttpRequestAndAnswer("cachedSerializedGroupMappingMap", () ->
groupResponseHolder);
// Return the second user request immediately
- expectHttpRequestAndAnswer(() -> userResponseHolder);
+ expectHttpRequestAndAnswer("cachedSerializedUserMap", () ->
userResponseHolder);
// Block the second group request so that it can be interrupted by stop()
final AtomicBoolean isInterrupted = new AtomicBoolean(false);
- expectHttpRequestAndAnswer(() -> {
- try {
- Thread.sleep(10_000);
- return groupResponseHolder;
- }
- catch (InterruptedException e) {
- isInterrupted.set(true);
- throw e;
- }
- });
+ expectHttpRequestAndAnswer(
+ "cachedSerializedGroupMappingMap",
+ () -> {
+ try {
+ Thread.sleep(10_000);
+ return groupResponseHolder;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw e;
+ }
+ }
+ );
replayAll();
@@ -183,21 +195,27 @@ public class
CoordinatorPollingBasicAuthorizerCacheManagerTest
verifyAll();
}
- private void expectHttpRequestAndAnswer(IAnswer<BytesFullResponseHolder>
responseHolder)
+ private void expectHttpRequestAndAnswer(String path,
IAnswer<BytesFullResponseHolder> responseHolder)
{
- try {
- EasyMock.expect(
- leaderClient.makeRequest(EasyMock.anyObject(), EasyMock.anyString())
- ).andReturn(request).once();
- EasyMock.expect(
- leaderClient.go(
- EasyMock.anyObject(),
- EasyMock.anyObject(BytesFullResponseHandler.class)
- )
- ).andAnswer(responseHolder).once();
- }
- catch (Exception e) {
- throw new RuntimeException(e);
- }
+ final String fullPath = StringUtils.format(
+ "/druid-ext/basic-security/authorization/db/%s/%s",
+ AUTHORIZER_NAME, path
+ );
+ serviceClient.expectAndRespond(
+ new RequestBuilder(HttpMethod.GET, fullPath),
+ new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK)
+ {
+ @Override
+ public ChannelBuffer getContent()
+ {
+ try {
+ return
ChannelBuffers.wrappedBuffer(responseHolder.answer().getContent());
+ }
+ catch (Throwable e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ );
}
}
diff --git a/extensions-core/druid-catalog/pom.xml
b/extensions-core/druid-catalog/pom.xml
index 729bac20861..9bd320c24a7 100644
--- a/extensions-core/druid-catalog/pom.xml
+++ b/extensions-core/druid-catalog/pom.xml
@@ -187,6 +187,11 @@
<artifactId>easymock</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-test</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.druid</groupId>
<artifactId>druid-processing</artifactId>
@@ -201,6 +206,24 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.druid</groupId>
+ <artifactId>druid-services</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.druid</groupId>
+ <artifactId>druid-indexing-service</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency><dependency>
+ <groupId>org.apache.druid</groupId>
+ <artifactId>druid-services</artifactId>
+ <version>${project.parent.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.druid</groupId>
<artifactId>druid-sql</artifactId>
diff --git
a/extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java
b/extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java
index 68bc21d571f..1c3a2be487b 100644
---
a/extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java
+++
b/extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java
@@ -21,67 +21,69 @@ package org.apache.druid.catalog.sync;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.catalog.http.CatalogResource;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.TableDefnRegistry;
import org.apache.druid.catalog.model.TableId;
import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.TableSpec;
import org.apache.druid.client.coordinator.Coordinator;
-import org.apache.druid.discovery.DruidLeaderClient;
-import org.apache.druid.guice.annotations.Json;
-import org.apache.druid.guice.annotations.Smile;
-import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.common.guava.FutureUtils;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.annotations.EscalatedGlobal;
import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.http.client.Request;
-import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
-import org.jboss.netty.handler.codec.http.HttpHeaders;
+import org.apache.druid.java.util.common.jackson.JacksonUtils;
+import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
+import org.apache.druid.rpc.IgnoreHttpResponseHandler;
+import org.apache.druid.rpc.RequestBuilder;
+import org.apache.druid.rpc.ServiceClient;
+import org.apache.druid.rpc.ServiceClientFactory;
+import org.apache.druid.rpc.ServiceLocator;
+import org.apache.druid.rpc.StandardRetryPolicy;
+import org.apache.druid.server.http.ServletResourceUtils;
import org.jboss.netty.handler.codec.http.HttpMethod;
-import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import javax.annotation.Nullable;
import javax.inject.Inject;
-import javax.ws.rs.core.MediaType;
-import java.io.IOException;
import java.util.Collections;
import java.util.List;
/**
- * Guice-injected client for the catalog update sync process. Requests
- * tables and schemas from the catalog component on the Coordinator.
- *
- * This class handles any recoverable error case. If this class throws
- * an exception, then something went very wrong and there is little the
- * caller can do to make things better. All the caller can do is try
- * again later and hope things improve.
+ * Used by Brokers and Overlord to sync the catalog by querying the
Coordinator.
+ * <p>
+ * Modeled after other {@link ServiceClient}-based classes such as {@code
CoordinatorClient}.
*/
public class CatalogClient implements CatalogSource
{
public static final String SCHEMA_SYNC_PATH = CatalogResource.ROOT_PATH +
CatalogResource.SCHEMA_SYNC;
public static final String TABLE_SYNC_PATH = CatalogResource.ROOT_PATH +
CatalogResource.TABLE_SYNC;
- private static final TypeReference<List<TableMetadata>>
LIST_OF_TABLE_METADATA_TYPE = new TypeReference<>() {};
- // Not strictly needed as a TypeReference, but doing so makes the code
simpler.
- private static final TypeReference<TableMetadata> TABLE_METADATA_TYPE = new
TypeReference<>() {};
+ private static final String TABLE_CREATE_PATH = CatalogResource.ROOT_PATH +
"/schemas/{schema}/tables/{name}";
- private final DruidLeaderClient coordClient;
- private final ObjectMapper smileMapper;
+ private final ServiceClient serviceClient;
+ private final ObjectMapper jsonMapper;
private final TableDefnRegistry tableRegistry;
@Inject
public CatalogClient(
- @Coordinator DruidLeaderClient coordClient,
- @Smile ObjectMapper smileMapper,
- @Json ObjectMapper jsonMapper
+ @EscalatedGlobal final ServiceClientFactory clientFactory,
+ @Coordinator final ServiceLocator serviceLocator,
+ ObjectMapper jsonMapper
)
{
- this.coordClient = coordClient;
- this.smileMapper = smileMapper;
+ this.jsonMapper = jsonMapper;
this.tableRegistry = new TableDefnRegistry(jsonMapper);
+ this.serviceClient = clientFactory.makeClient(
+ NodeRole.COORDINATOR.getJsonName(),
+ serviceLocator,
+ StandardRetryPolicy.builder().maxAttempts(6).build()
+ );
}
@Override
public List<TableMetadata> tablesForSchema(String dbSchema)
{
- String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{schema}",
StringUtils.urlEncode(dbSchema));
- List<TableMetadata> results = send(url, LIST_OF_TABLE_METADATA_TYPE);
+ List<TableMetadata> results = getResult(fetchTablesForSchema(dbSchema));
// Not found for a list is an empty list.
return results == null ? Collections.emptyList() : results;
@@ -90,9 +92,7 @@ public class CatalogClient implements CatalogSource
@Override
public TableMetadata table(TableId id)
{
- String url = StringUtils.replace(TABLE_SYNC_PATH, "{schema}",
StringUtils.urlEncode(id.schema()));
- url = StringUtils.replace(url, "{name}", StringUtils.urlEncode(id.name()));
- return send(url, TABLE_METADATA_TYPE);
+ return getResult(fetchTableForId(id));
}
@Override
@@ -103,46 +103,87 @@ public class CatalogClient implements CatalogSource
}
/**
- * Send the update. Exceptions are "unexpected": they should never occur in a
- * working system. If they occur, something is broken.
+ * Creates a table for the given {@link TableId} and {@link TableSpec}.
+ * If a table already exists for this id, it is overwritten.
+ * <p>
+ * This method is currently used only in tests.
+ */
+ public void createTable(TableId tableId, TableSpec tableSpec)
+ {
+ getResult(postCreateTable(tableId, tableSpec));
+ }
+
+ /**
+ * Posts the given table spec to the Coordinator.
+ * <p>
+ * API: {@code POST
/druid/coordinator/v1/catalog/schemas/{schema}/tables/{name}}
+ */
+ private ListenableFuture<Void> postCreateTable(TableId tableId, TableSpec
tableSpec)
+ {
+ String path = StringUtils.replace(TABLE_CREATE_PATH, "{schema}",
StringUtils.urlEncode(tableId.schema()));
+ path = StringUtils.replace(path, "{name}",
StringUtils.urlEncode(tableId.name()));
+
+ return serviceClient.asyncRequest(
+ new RequestBuilder(HttpMethod.POST, path)
+ .jsonContent(jsonMapper, tableSpec),
+ IgnoreHttpResponseHandler.INSTANCE
+ );
+ }
+
+ /**
+ * Fetches the metadata of a table from the Coordinator.
+ * <p>
+ * API: {@code GET
/druid/coordinator/v1/catalog/sync/schemas/{schema}/{name}}
+ */
+ private ListenableFuture<TableMetadata> fetchTableForId(TableId id)
+ {
+ String path = StringUtils.replace(TABLE_SYNC_PATH, "{schema}",
StringUtils.urlEncode(id.schema()));
+ path = StringUtils.replace(path, "{name}",
StringUtils.urlEncode(id.name()));
+
+ return FutureUtils.transform(
+ serviceClient.asyncRequest(
+ new RequestBuilder(HttpMethod.GET, path),
+ new BytesFullResponseHandler()
+ ),
+ holder -> JacksonUtils.readValue(jsonMapper, holder.getContent(),
TableMetadata.class)
+ );
+ }
+
+ /**
+ * Fetches the metadata of all tables in a schema from the Coordinator.
+ * <p>
+ * API: {@code GET /druid/coordinator/v1/catalog/sync/schemas/{schema}}
+ */
+ private ListenableFuture<List<TableMetadata>> fetchTablesForSchema(String
schemaName)
+ {
+ final String path = StringUtils.replace(
+ SCHEMA_SYNC_PATH,
+ "{schema}",
+ StringUtils.urlEncode(schemaName)
+ );
+
+ return FutureUtils.transform(
+ serviceClient.asyncRequest(
+ new RequestBuilder(HttpMethod.GET, path),
+ new BytesFullResponseHandler()
+ ),
+ holder -> JacksonUtils.readValue(jsonMapper, holder.getContent(), new
TypeReference<>() {})
+ );
+ }
+
+ /**
+ * Gets the result from the given future.
*
- * @return the requested update, or null if the item was not found in the
- * catalog.
+ * @return null if the API returns a 404 not found status.
*/
- private <T> T send(String url, TypeReference<T> typeRef)
+ @Nullable
+ private <T> T getResult(ListenableFuture<T> future)
{
- final Request request;
- try {
- request = coordClient.makeRequest(HttpMethod.GET, url)
- .addHeader(HttpHeaders.Names.ACCEPT, MediaType.APPLICATION_JSON);
- }
- catch (IOException e) {
- throw new ISE("Cannot create catalog sync request");
- }
- final StringFullResponseHolder responseHolder;
- try {
- responseHolder = coordClient.go(request);
- }
- catch (IOException e) {
- throw new ISE(e, "Failed to send catalog sync");
- }
- catch (InterruptedException e1) {
- // Treat as a not-found: the only way this exception should occur
- // is during shutdown.
- return null;
- }
- if (responseHolder.getStatus().getCode() ==
HttpResponseStatus.NOT_FOUND.getCode()) {
- // Not found means the item disappeared. Returning null means "not
found".
- return null;
- }
- if (responseHolder.getStatus().getCode() !=
HttpResponseStatus.OK.getCode()) {
- throw new ISE("Unexpected status from catalog sync: " +
responseHolder.getStatus());
- }
try {
- return smileMapper.readValue(responseHolder.getContent(), typeRef);
+ return FutureUtils.getUnchecked(future, true);
}
- catch (IOException e) {
- throw new ISE(e, "Could not decode the JSON response from catalog
sync.");
+ catch (Exception e) {
+ return ServletResourceUtils.getDefaultValueIfCauseIs404ElseThrow(e, ()
-> null);
}
}
}
diff --git
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
new file mode 100644
index 00000000000..d44f3ad6593
--- /dev/null
+++
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.catalog.compact;
+
+import org.apache.druid.catalog.guice.CatalogClientModule;
+import org.apache.druid.catalog.guice.CatalogCoordinatorModule;
+import org.apache.druid.catalog.model.TableId;
+import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.table.TableBuilder;
+import org.apache.druid.catalog.sync.CatalogClient;
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.indexing.compact.CompactionSupervisorSpec;
+import org.apache.druid.indexing.overlord.Segments;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.rpc.UpdateResponse;
+import org.apache.druid.server.coordinator.CatalogDataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.ClusterCompactionConfig;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedClusterApis;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.indexing.Resources;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.apache.druid.timeline.DataSegment;
+import org.joda.time.Period;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class CatalogCompactionTest extends EmbeddedClusterTestBase
+{
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator()
+ .addProperty("druid.manager.segments.useIncrementalCache", "always");
+ private final EmbeddedBroker broker = new EmbeddedBroker()
+ .addProperty("druid.catalog.client.pollingPeriod", "100");
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addExtension(CatalogClientModule.class)
+ .addExtension(CatalogCoordinatorModule.class)
+ .addServer(coordinator)
+ .addServer(overlord)
+ .addServer(broker)
+ .addServer(new EmbeddedIndexer())
+ .addServer(new EmbeddedHistorical());
+ }
+
+ @Test
+ public void test_ingestDayGranularity_andCompactToMonthGranularity()
+ {
+ // Ingest data at DAY granularity and verify
+ runIngestionAtDayGranularity();
+ List<DataSegment> segments = List.copyOf(
+ overlord.bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+ );
+ Assertions.assertEquals(10, segments.size());
+ segments.forEach(
+ segment ->
Assertions.assertTrue(Granularities.DAY.isAligned(segment.getInterval()))
+ );
+
+ final CatalogClient catalogClient =
broker.bindings().getInstance(CatalogClient.class);
+
+ // Create a catalog entry to have MONTH granularity for the datasource
+ final TableId tableId = new TableId("druid", dataSource);
+ catalogClient.createTable(
+ tableId,
+ TableBuilder.datasource(dataSource, "P1M").buildSpec()
+ );
+
+ final TableMetadata metadata = catalogClient.table(tableId);
+ Assertions.assertEquals(
+ StringUtils.format("\"druid\".\"%s\"", dataSource),
+ metadata.sqlName()
+ );
+ Assertions.assertEquals(
+ TableBuilder.datasource(dataSource, "P1M").buildSpec(),
+ metadata.spec()
+ );
+
+ enableCompactionSupervisor();
+
+ // Create a catalog compaction config
+ CatalogDataSourceCompactionConfig compactionConfig =
+ new CatalogDataSourceCompactionConfig(dataSource, null, Period.ZERO,
null, null, null, null);
+
+ final CompactionSupervisorSpec compactionSupervisor
+ = new CompactionSupervisorSpec(compactionConfig, false, null);
+ cluster.callApi().onLeaderOverlord(o ->
o.postSupervisor(compactionSupervisor));
+
+ // Wait for compaction to finish
+ overlord.latchableEmitter().waitForEvent(
+ event -> event.hasMetricName("task/run/time")
+ .hasDimension("taskType", "compact")
+ );
+
+ // Verify that segments are now compacted to MONTH granularity
+ segments = List.copyOf(
+ overlord.bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+ );
+ Assertions.assertEquals(1, segments.size());
+ Assertions.assertTrue(
+ Granularities.MONTH.isAligned(segments.get(0).getInterval())
+ );
+ }
+
+ private void runIngestionAtDayGranularity()
+ {
+ final String taskId = IdUtils.getRandomId();
+ final Object task = createIndexTaskForInlineData(
+ taskId,
+ StringUtils.replace(Resources.CSV_DATA_10_DAYS, "\n", "\\n")
+ );
+
+ cluster.callApi().onLeaderOverlord(o -> o.runTask(taskId, task));
+ cluster.callApi().waitForTaskToSucceed(taskId, overlord);
+ }
+
+ private Object createIndexTaskForInlineData(String taskId, String
inlineDataCsv)
+ {
+ return EmbeddedClusterApis.createTaskFromPayload(
+ taskId,
+ StringUtils.format(Resources.INDEX_TASK_PAYLOAD_WITH_INLINE_DATA,
inlineDataCsv, dataSource)
+ );
+ }
+
+ private void enableCompactionSupervisor()
+ {
+ final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
+ o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig(1.0,
10, null, true, null))
+ );
+ Assertions.assertTrue(updateResponse.isSuccess());
+ }
+}
diff --git
a/quidem-ut/src/main/java/org/apache/druid/quidem/ExposedAsBrokerQueryComponentSupplierWrapper.java
b/quidem-ut/src/main/java/org/apache/druid/quidem/ExposedAsBrokerQueryComponentSupplierWrapper.java
index 0a2e496b09a..ede29251aec 100644
---
a/quidem-ut/src/main/java/org/apache/druid/quidem/ExposedAsBrokerQueryComponentSupplierWrapper.java
+++
b/quidem-ut/src/main/java/org/apache/druid/quidem/ExposedAsBrokerQueryComponentSupplierWrapper.java
@@ -41,7 +41,6 @@ import org.apache.druid.curator.discovery.DiscoveryModule;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.guice.AnnouncerModule;
import org.apache.druid.guice.BrokerServiceModule;
-import org.apache.druid.guice.CoordinatorDiscoveryModule;
import org.apache.druid.guice.ExpressionModule;
import org.apache.druid.guice.ExtensionsModule;
import org.apache.druid.guice.JacksonConfigManagerModule;
@@ -186,7 +185,6 @@ public class ExposedAsBrokerQueryComponentSupplierWrapper
extends QueryComponent
new MetadataConfigModule(),
new DerbyMetadataStorageDruidModule(),
new JacksonConfigManagerModule(),
- new CoordinatorDiscoveryModule(),
new LocalDataStorageDruidModule(),
new TombstoneDataStorageModule(),
new JavaScriptModule(),
diff --git
a/server/src/main/java/org/apache/druid/discovery/DruidLeaderClient.java
b/server/src/main/java/org/apache/druid/discovery/DruidLeaderClient.java
deleted file mode 100644
index 28e48b29aa4..00000000000
--- a/server/src/main/java/org/apache/druid/discovery/DruidLeaderClient.java
+++ /dev/null
@@ -1,307 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.druid.discovery;
-
-import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
-import org.apache.druid.concurrent.LifecycleLock;
-import org.apache.druid.java.util.common.IOE;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.common.RE;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
-import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
-import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.Request;
-import org.apache.druid.java.util.http.client.response.FullResponseHolder;
-import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
-import
org.apache.druid.java.util.http.client.response.StringFullResponseHandler;
-import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
-import org.jboss.netty.channel.ChannelException;
-import org.jboss.netty.handler.codec.http.HttpMethod;
-import org.jboss.netty.handler.codec.http.HttpResponseStatus;
-
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.charset.StandardCharsets;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * This class facilitates interaction with Coordinator/Overlord leader nodes.
Instance of this class is injected
- * via Guice with annotations @Coordinator or @IndexingService .
- * Usage:
- * Request request = druidLeaderClient.makeRequest(HttpMethod, requestPath)
- * request.setXXX(..)
- * FullResponseHolder responseHolder = druidLeaderClient.go(request)
- */
-public class DruidLeaderClient
-{
- private final Logger log = new Logger(DruidLeaderClient.class);
-
- private static final int MAX_RETRIES = 5;
-
- private final HttpClient httpClient;
- private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
- private final NodeRole nodeRoleToWatch;
-
- private final String leaderRequestPath;
-
- private final LifecycleLock lifecycleLock = new LifecycleLock();
- private DruidNodeDiscovery druidNodeDiscovery;
- private final AtomicReference<String> currentKnownLeader = new
AtomicReference<>();
-
- public DruidLeaderClient(
- HttpClient httpClient,
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider,
- NodeRole nodeRoleToWatch,
- String leaderRequestPath
- )
- {
- this.httpClient = httpClient;
- this.druidNodeDiscoveryProvider = druidNodeDiscoveryProvider;
- this.nodeRoleToWatch = nodeRoleToWatch;
- this.leaderRequestPath = leaderRequestPath;
- }
-
- @LifecycleStart
- public void start()
- {
- if (!lifecycleLock.canStart()) {
- throw new ISE("can't start.");
- }
-
- try {
- druidNodeDiscovery =
druidNodeDiscoveryProvider.getForNodeRole(nodeRoleToWatch);
- lifecycleLock.started();
- log.debug("Started.");
- }
- finally {
- lifecycleLock.exitStart();
- }
- }
-
- @LifecycleStop
- public void stop()
- {
- if (!lifecycleLock.canStop()) {
- throw new ISE("can't stop.");
- }
-
- log.debug("Stopped.");
- }
-
- /**
- * Make a Request object aimed at the leader. Throws IOException if the
leader cannot be located.
- */
- public Request makeRequest(HttpMethod httpMethod, String urlPath) throws
IOException
- {
- Preconditions.checkState(lifecycleLock.awaitStarted(1,
TimeUnit.MILLISECONDS));
- return new Request(httpMethod, new URL(StringUtils.format("%s%s",
getCurrentKnownLeader(true), urlPath)));
- }
-
- /**
- * Executes a Request object aimed at the leader. Throws IOException if the
leader cannot be located.
- * Internal retries with cache invalidation are attempted if 503/504
response is received.
- *
- * @param request
- * @throws IOException
- * @throws InterruptedException
- */
- public StringFullResponseHolder go(Request request) throws IOException,
InterruptedException
- {
- return go(request, new StringFullResponseHandler(StandardCharsets.UTF_8));
- }
-
- /**
- * Executes a Request object aimed at the leader. Throws IOException if the
leader cannot be located.
- * Internal retries with cache invalidation are attempted if 503/504
response is received.
- *
- * @param request
- * @param responseHandler
- * @throws IOException
- * @throws InterruptedException
- */
- public <T, H extends FullResponseHolder<T>> H go(
- Request request,
- HttpResponseHandler<H, H> responseHandler
- )
- throws IOException, InterruptedException
- {
- Preconditions.checkState(lifecycleLock.awaitStarted(1,
TimeUnit.MILLISECONDS));
- for (int counter = 0; counter < MAX_RETRIES; counter++) {
-
- final H fullResponseHolder;
-
- try {
- try {
- fullResponseHolder = httpClient.go(request, responseHandler).get();
- }
- catch (ExecutionException e) {
- // Unwrap IOExceptions and ChannelExceptions, re-throw others
- Throwables.propagateIfInstanceOf(e.getCause(), IOException.class);
- Throwables.propagateIfInstanceOf(e.getCause(),
ChannelException.class);
- throw new RE(e, "HTTP request to[%s] failed", request.getUrl());
- }
- }
- catch (IOException | ChannelException ex) {
- // can happen if the node is stopped.
- log.warn(ex, "Request[%s] failed.", request.getUrl());
- request = getNewRequestUrlInvalidatingCache(request);
- continue;
- }
- HttpResponseStatus responseStatus =
fullResponseHolder.getResponse().getStatus();
- if (HttpResponseStatus.TEMPORARY_REDIRECT.equals(responseStatus)) {
- String redirectUrlStr =
fullResponseHolder.getResponse().headers().get("Location");
- if (redirectUrlStr == null) {
- throw new IOE("No redirect location is found in response from
url[%s].", request.getUrl());
- }
-
- log.info("Request[%s] received redirect response to location [%s].",
request.getUrl(), redirectUrlStr);
-
- final URL redirectUrl;
- try {
- redirectUrl = new URL(redirectUrlStr);
- }
- catch (MalformedURLException ex) {
- throw new IOE(
- ex,
- "Malformed redirect location is found in response from url[%s],
new location[%s].",
- request.getUrl(),
- redirectUrlStr
- );
- }
-
- //update known leader location
- currentKnownLeader.set(StringUtils.format(
- "%s://%s:%s",
- redirectUrl.getProtocol(),
- redirectUrl.getHost(),
- redirectUrl.getPort()
- ));
-
- request = ClientUtils.withUrl(request, redirectUrl);
- } else if (HttpResponseStatus.SERVICE_UNAVAILABLE.equals(responseStatus)
- || HttpResponseStatus.GATEWAY_TIMEOUT.equals(responseStatus))
{
- log.warn(
- "Request[%s] received a %s response. Attempt %s/%s",
- request.getUrl(),
- responseStatus,
- counter + 1,
- MAX_RETRIES
- );
- request = getNewRequestUrlInvalidatingCache(request);
- } else {
- return fullResponseHolder;
- }
- }
-
- throw new IOE("Retries exhausted, couldn't fulfill request to [%s].",
request.getUrl());
- }
-
- public String findCurrentLeader()
- {
- Preconditions.checkState(lifecycleLock.awaitStarted(1,
TimeUnit.MILLISECONDS));
- final StringFullResponseHolder responseHolder;
- try {
- responseHolder = go(makeRequest(HttpMethod.GET, leaderRequestPath));
- }
- catch (Exception ex) {
- throw new ISE(ex, "Couldn't find leader.");
- }
-
- if (responseHolder.getStatus().getCode() == 200) {
- String leaderUrl = responseHolder.getContent();
-
- //verify this is valid url
- try {
- URL validatedUrl = new URL(leaderUrl);
- currentKnownLeader.set(leaderUrl);
-
- // validatedUrl.toString() is returned instead of leaderUrl or else
teamcity build fails because of breaking
- // the rule of ignoring new URL(leaderUrl) object.
- return validatedUrl.toString();
- }
- catch (MalformedURLException ex) {
- log.error(ex, "Received malformed leader url[%s].", leaderUrl);
- }
- }
-
- throw new ISE(
- "Couldn't find leader, failed response status is [%s] and content
[%s].",
- responseHolder.getStatus().getCode(),
- responseHolder.getContent()
- );
- }
-
- private String getCurrentKnownLeader(final boolean cached) throws IOException
- {
- final String leader = currentKnownLeader.accumulateAndGet(
- null,
- (current, given) -> current == null || !cached ?
ClientUtils.pickOneHost(druidNodeDiscovery) : current
- );
-
- if (leader == null) {
- throw new IOE(
- "A leader node could not be found for [%s] service. "
- + "Check logs of service [%s] to confirm it is healthy.",
- nodeRoleToWatch, nodeRoleToWatch
- );
- } else {
- return leader;
- }
- }
-
- private Request getNewRequestUrlInvalidatingCache(Request oldRequest) throws
IOException
- {
- try {
- Request newRequest;
- if (oldRequest.getUrl().getQuery() == null) {
- newRequest = ClientUtils.withUrl(
- oldRequest,
- new URL(StringUtils.format("%s%s", getCurrentKnownLeader(false),
oldRequest.getUrl().getPath()))
- );
- } else {
- newRequest = ClientUtils.withUrl(
- oldRequest,
- new URL(StringUtils.format(
- "%s%s?%s",
- getCurrentKnownLeader(false),
- oldRequest.getUrl().getPath(),
- oldRequest.getUrl().getQuery()
- ))
- );
- }
- return newRequest;
- }
- catch (MalformedURLException e) {
- // Not an IOException; this is our own fault.
- throw new ISE(
- e,
- "failed to build url with path[%] and query string [%s].",
- oldRequest.getUrl().getPath(),
- oldRequest.getUrl().getQuery()
- );
- }
- }
-}
diff --git
a/server/src/main/java/org/apache/druid/guice/CoordinatorDiscoveryModule.java
b/server/src/main/java/org/apache/druid/guice/CoordinatorDiscoveryModule.java
deleted file mode 100644
index d9f083767df..00000000000
---
a/server/src/main/java/org/apache/druid/guice/CoordinatorDiscoveryModule.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.druid.guice;
-
-import com.google.inject.Binder;
-import com.google.inject.Module;
-import com.google.inject.Provides;
-import org.apache.druid.client.coordinator.Coordinator;
-import org.apache.druid.client.coordinator.CoordinatorSelectorConfig;
-import org.apache.druid.discovery.DruidLeaderClient;
-import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
-import org.apache.druid.discovery.NodeRole;
-import org.apache.druid.guice.annotations.EscalatedGlobal;
-import org.apache.druid.java.util.http.client.HttpClient;
-
-/**
- */
-public class CoordinatorDiscoveryModule implements Module
-{
- @Override
- public void configure(Binder binder)
- {
- JsonConfigProvider.bind(binder, "druid.selectors.coordinator",
CoordinatorSelectorConfig.class);
- }
-
- @Provides
- @Coordinator
- @ManageLifecycle
- public DruidLeaderClient getLeaderHttpClient(
- @EscalatedGlobal HttpClient httpClient,
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider
- )
- {
- return new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.COORDINATOR,
- "/druid/coordinator/v1/leader"
- );
- }
-}
diff --git
a/server/src/main/java/org/apache/druid/initialization/CoreInjectorBuilder.java
b/server/src/main/java/org/apache/druid/initialization/CoreInjectorBuilder.java
index caef8be64b1..602fe4ec3d2 100644
---
a/server/src/main/java/org/apache/druid/initialization/CoreInjectorBuilder.java
+++
b/server/src/main/java/org/apache/druid/initialization/CoreInjectorBuilder.java
@@ -26,7 +26,6 @@ import org.apache.druid.discovery.NodeRole;
import org.apache.druid.guice.AnnouncerModule;
import org.apache.druid.guice.BuiltInTypesModule;
import org.apache.druid.guice.CatalogCoreModule;
-import org.apache.druid.guice.CoordinatorDiscoveryModule;
import org.apache.druid.guice.DruidInjectorBuilder;
import org.apache.druid.guice.DruidSecondaryModule;
import org.apache.druid.guice.ExpressionModule;
@@ -122,7 +121,6 @@ public class CoreInjectorBuilder extends
DruidInjectorBuilder
new MetadataConfigModule(),
new DerbyMetadataStorageDruidModule(),
new JacksonConfigManagerModule(),
- new CoordinatorDiscoveryModule(),
new LocalDataStorageDruidModule(),
new TombstoneDataStorageModule(),
new JavaScriptModule(),
diff --git
a/server/src/main/java/org/apache/druid/query/lookup/LookupReferencesManager.java
b/server/src/main/java/org/apache/druid/query/lookup/LookupReferencesManager.java
index cef7d9eaa29..94c333122b7 100644
---
a/server/src/main/java/org/apache/druid/query/lookup/LookupReferencesManager.java
+++
b/server/src/main/java/org/apache/druid/query/lookup/LookupReferencesManager.java
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
-import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Inject;
@@ -41,10 +40,9 @@ import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.rpc.HttpResponseException;
+import org.apache.druid.server.http.ServletResourceUtils;
import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
import org.apache.druid.server.metrics.DataSourceTaskIdHolder;
-import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import javax.annotation.Nullable;
import java.io.File;
@@ -475,19 +473,13 @@ public class LookupReferencesManager implements
LookupExtractorFactoryContainerP
return coordinatorClient.fetchLookupsForTierSync(tier);
}
catch (Exception e) {
- Throwable rootCause = Throwables.getRootCause(e);
- if (rootCause instanceof HttpResponseException) {
- final HttpResponseException httpException = (HttpResponseException)
rootCause;
- if
(httpException.getResponse().getStatus().equals(HttpResponseStatus.NOT_FOUND)) {
- LOG.info(
- "No lookups found for tier [%s], status [%s]",
- tier,
- httpException.getResponse().getStatus()
- );
- return null;
- }
- }
- throw e;
+ return ServletResourceUtils.getDefaultValueIfCauseIs404ElseThrow(
+ e,
+ () -> {
+ LOG.info("No lookups found for tier[%s], status [404, Not Found]",
tier);
+ return null;
+ }
+ );
}
}
diff --git
a/server/src/main/java/org/apache/druid/server/http/ServletResourceUtils.java
b/server/src/main/java/org/apache/druid/server/http/ServletResourceUtils.java
index 04465b677bf..91c59304bfb 100644
---
a/server/src/main/java/org/apache/druid/server/http/ServletResourceUtils.java
+++
b/server/src/main/java/org/apache/druid/server/http/ServletResourceUtils.java
@@ -26,6 +26,8 @@ import org.apache.druid.error.ErrorResponse;
import org.apache.druid.error.InternalServerError;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.rpc.HttpResponseException;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import javax.annotation.Nullable;
import javax.ws.rs.core.MediaType;
@@ -89,4 +91,25 @@ public class ServletResourceUtils
);
}
}
+
+ /**
+ * Returns the given default value if the root cause of the exception is an
+ * {@link HttpResponseException} with status code {@link
HttpResponseStatus#NOT_FOUND}.
+ * Otherwise, re-throws the given exception wrapped in a {@link
RuntimeException}.
+ */
+ public static <T> T getDefaultValueIfCauseIs404ElseThrow(
+ Exception e,
+ Supplier<T> defaultValueSupplier
+ )
+ {
+ Throwable rootCause = Throwables.getRootCause(e);
+ if (rootCause instanceof HttpResponseException) {
+ final HttpResponseException httpException = (HttpResponseException)
rootCause;
+ if
(httpException.getResponse().getStatus().equals(HttpResponseStatus.NOT_FOUND)) {
+ return defaultValueSupplier.get();
+ }
+ }
+
+ throw new RuntimeException(e);
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/discovery/DruidLeaderClientTest.java
b/server/src/test/java/org/apache/druid/discovery/DruidLeaderClientTest.java
deleted file mode 100644
index 4d9870c6e6e..00000000000
--- a/server/src/test/java/org/apache/druid/discovery/DruidLeaderClientTest.java
+++ /dev/null
@@ -1,338 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.druid.discovery;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.util.concurrent.ListenableFutureTask;
-import com.google.inject.Binder;
-import com.google.inject.Inject;
-import com.google.inject.Injector;
-import com.google.inject.Key;
-import com.google.inject.Module;
-import com.google.inject.name.Named;
-import com.google.inject.name.Names;
-import com.google.inject.servlet.GuiceFilter;
-import org.apache.druid.guice.GuiceInjectors;
-import org.apache.druid.guice.Jerseys;
-import org.apache.druid.guice.JsonConfigProvider;
-import org.apache.druid.guice.LazySingleton;
-import org.apache.druid.guice.LifecycleModule;
-import org.apache.druid.guice.annotations.Self;
-import org.apache.druid.initialization.Initialization;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.Request;
-import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
-import org.apache.druid.server.DruidNode;
-import org.apache.druid.server.initialization.BaseJettyTest;
-import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
-import org.easymock.EasyMock;
-import org.eclipse.jetty.server.Handler;
-import org.eclipse.jetty.server.Server;
-import org.eclipse.jetty.server.handler.HandlerList;
-import org.eclipse.jetty.servlet.DefaultServlet;
-import org.eclipse.jetty.servlet.ServletContextHandler;
-import org.eclipse.jetty.servlet.ServletHolder;
-import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
-import org.jboss.netty.handler.codec.http.HttpMethod;
-import org.jboss.netty.handler.codec.http.HttpResponseStatus;
-import org.jboss.netty.handler.codec.http.HttpVersion;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.mockito.ArgumentMatchers;
-import org.mockito.Mockito;
-
-import javax.ws.rs.GET;
-import javax.ws.rs.POST;
-import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
-import java.io.IOException;
-import java.net.URI;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-
-/**
- */
-@SuppressWarnings("DoNotMock")
-public class DruidLeaderClientTest extends BaseJettyTest
-{
- @Rule
- public ExpectedException expectedException = ExpectedException.none();
-
- private DiscoveryDruidNode discoveryDruidNode;
- private HttpClient httpClient;
-
- @Override
- protected Injector setupInjector()
- {
- final DruidNode node = new DruidNode("test", "localhost", false, null,
null, true, false);
- discoveryDruidNode = new DiscoveryDruidNode(node, NodeRole.PEON,
ImmutableMap.of());
-
- Injector injector = Initialization.makeInjectorWithModules(
- GuiceInjectors.makeStartupInjector(), ImmutableList.<Module>of(
- new Module()
- {
- @Override
- public void configure(Binder binder)
- {
- JsonConfigProvider.bindInstance(
- binder,
- Key.get(DruidNode.class, Self.class),
- node
- );
-
binder.bind(Integer.class).annotatedWith(Names.named("port")).toInstance(node.getPlaintextPort());
-
binder.bind(JettyServerInitializer.class).to(TestJettyServerInitializer.class).in(LazySingleton.class);
- Jerseys.addResource(binder, SimpleResource.class);
- LifecycleModule.register(binder, Server.class);
- }
- }
- )
- );
- httpClient = injector.getInstance(ClientHolder.class).getClient();
- return injector;
- }
-
- @Test
- public void testSimple() throws Exception
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
- EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(
- ImmutableList.of(discoveryDruidNode)
- );
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery);
-
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider);
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- Request request = druidLeaderClient.makeRequest(HttpMethod.POST,
"/simple/direct");
- request.setContent("hello".getBytes(StandardCharsets.UTF_8));
- Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
- }
-
- @Test
- public void testNoLeaderFound() throws Exception
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
-
EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(ImmutableList.of());
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery);
-
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider);
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- expectedException.expect(IOException.class);
- expectedException.expectMessage(
- "A leader node could not be found for [PEON] service. "
- + "Check logs of service [PEON] to confirm it is healthy.");
- druidLeaderClient.makeRequest(HttpMethod.POST, "/simple/direct");
- }
-
- @Test
- public void testRedirection() throws Exception
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
- EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(
- ImmutableList.of(discoveryDruidNode)
- );
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery);
-
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider);
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- Request request = druidLeaderClient.makeRequest(HttpMethod.POST,
"/simple/redirect");
- request.setContent("hello".getBytes(StandardCharsets.UTF_8));
- Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
- }
-
- @Test
- public void testServerFailureAndRedirect() throws Exception
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
- DiscoveryDruidNode dummyNode = new DiscoveryDruidNode(
- new DruidNode("test", "dummyhost", false, 64231, null, true, false),
- NodeRole.PEON,
- ImmutableMap.of()
- );
-
EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(ImmutableList.of(dummyNode));
-
EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(ImmutableList.of(discoveryDruidNode));
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery).anyTimes();
-
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider);
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- Request request = druidLeaderClient.makeRequest(HttpMethod.POST,
"/simple/redirect");
- request.setContent("hello".getBytes(StandardCharsets.UTF_8));
- Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
- }
-
- @Test
- public void test503ResponseFromServerAndCacheRefresh() throws Exception
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
- // Should be called twice. Second time is when we refresh the cache since
we get 503 in the first request
-
EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(ImmutableList.of(discoveryDruidNode)).times(2);
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery).anyTimes();
-
- ListenableFutureTask task =
EasyMock.createMock(ListenableFutureTask.class);
- EasyMock.expect(task.get()).andReturn(new StringFullResponseHolder(new
DefaultHttpResponse(
- HttpVersion.HTTP_1_1,
- HttpResponseStatus.SERVICE_UNAVAILABLE
- ), Charset.defaultCharset()));
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider, task);
-
- HttpClient spyHttpClient = Mockito.spy(this.httpClient);
- // Override behavior for the first call only
-
Mockito.doReturn(task).doCallRealMethod().when(spyHttpClient).go(ArgumentMatchers.any(),
ArgumentMatchers.any());
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- spyHttpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- Request request = druidLeaderClient.makeRequest(HttpMethod.POST,
"/simple/direct");
- request.setContent("hello".getBytes(StandardCharsets.UTF_8));
- Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
- EasyMock.verify(druidNodeDiscovery);
- }
-
- @Test
- public void testFindCurrentLeader()
- {
- DruidNodeDiscovery druidNodeDiscovery =
EasyMock.createMock(DruidNodeDiscovery.class);
- EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(
- ImmutableList.of(discoveryDruidNode)
- );
-
- DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
-
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery);
-
- EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider);
-
- DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
- httpClient,
- druidNodeDiscoveryProvider,
- NodeRole.PEON,
- "/simple/leader"
- );
- druidLeaderClient.start();
-
- Assert.assertEquals("http://localhost:1234/";,
druidLeaderClient.findCurrentLeader());
- }
-
- static class TestJettyServerInitializer implements JettyServerInitializer
- {
- @Override
- public void initialize(Server server, Injector injector)
- {
- final ServletContextHandler root = new
ServletContextHandler(ServletContextHandler.SESSIONS);
- root.addServlet(new ServletHolder(new DefaultServlet()), "/*");
- root.addFilter(GuiceFilter.class, "/*", null);
-
- final HandlerList handlerList = new HandlerList();
- handlerList.setHandlers(new Handler[]{root});
- server.setHandler(handlerList);
- }
- }
-
- @Path("/simple")
- public static class SimpleResource
- {
- private final int port;
-
- @Inject
- public SimpleResource(@Named("port") int port)
- {
- this.port = port;
- }
-
- @POST
- @Path("/direct")
- @Produces(MediaType.APPLICATION_JSON)
- public Response direct(String input)
- {
- if ("hello".equals(input)) {
- return Response.ok("hello").build();
- } else {
- return Response.serverError().build();
- }
- }
-
- @POST
- @Path("/redirect")
- @Produces(MediaType.APPLICATION_JSON)
- public Response redirecting() throws Exception
- {
- return Response.temporaryRedirect(new
URI(StringUtils.format("http://localhost:%s/simple/direct";, port))).build();
- }
-
- @GET
- @Path("/leader")
- @Produces(MediaType.APPLICATION_JSON)
- public Response leader()
- {
- return Response.ok("http://localhost:1234/";).build();
- }
- }
-}
diff --git
a/server/src/test/java/org/apache/druid/server/http/ServletResourceUtilsTest.java
b/server/src/test/java/org/apache/druid/server/http/ServletResourceUtilsTest.java
index 0af66bf0df5..ddbc3d4376f 100644
---
a/server/src/test/java/org/apache/druid/server/http/ServletResourceUtilsTest.java
+++
b/server/src/test/java/org/apache/druid/server/http/ServletResourceUtilsTest.java
@@ -23,11 +23,18 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.error.ErrorResponse;
import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.ISE;
+import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
+import org.apache.druid.rpc.HttpResponseException;
import org.hamcrest.MatcherAssert;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.jboss.netty.handler.codec.http.HttpVersion;
import org.junit.Assert;
import org.junit.Test;
import javax.ws.rs.core.Response;
+import java.nio.charset.StandardCharsets;
public class ServletResourceUtilsTest
{
@@ -62,4 +69,37 @@ public class ServletResourceUtilsTest
DruidExceptionMatcher.invalidInput().expectMessageIs("Invalid value of
[inputKey]")
);
}
+
+ @Test
+ public void test_getDefaultValueIfCauseIs404ElseThrow()
+ {
+ final StringFullResponseHolder notFoundResponseHolder = new
StringFullResponseHolder(
+ new DefaultHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.NOT_FOUND),
+ StandardCharsets.UTF_8
+ );
+ Assert.assertEquals(
+ "abc",
+ ServletResourceUtils.getDefaultValueIfCauseIs404ElseThrow(
+ new ISE(new HttpResponseException(notFoundResponseHolder), ""),
+ () -> "abc"
+ )
+ );
+
+ final StringFullResponseHolder badRequestResponseHolder = new
StringFullResponseHolder(
+ new DefaultHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.BAD_REQUEST),
+ StandardCharsets.UTF_8
+ );
+ Assert.assertThrows(
+ RuntimeException.class,
+ () -> ServletResourceUtils.getDefaultValueIfCauseIs404ElseThrow(
+ new ISE(new HttpResponseException(badRequestResponseHolder), ""),
+ () -> "abc"
+ )
+ );
+
+ Assert.assertThrows(
+ RuntimeException.class,
+ () -> ServletResourceUtils.getDefaultValueIfCauseIs404ElseThrow(new
ISE(""), () -> "abc")
+ );
+ }
}
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/ServerReferenceHolder.java
b/services/src/test/java/org/apache/druid/testing/embedded/ServerReferenceHolder.java
index d36d85693b1..2e878621afe 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/ServerReferenceHolder.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/ServerReferenceHolder.java
@@ -21,6 +21,7 @@ package org.apache.druid.testing.embedded;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Inject;
+import com.google.inject.Injector;
import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
@@ -87,6 +88,9 @@ public final class ServerReferenceHolder implements
ServerReferencesProvider
@Json
private ObjectMapper jsonMapper;
+ @Inject
+ private Injector injector;
+
@Override
public DruidNode selfNode()
{
@@ -158,4 +162,10 @@ public final class ServerReferenceHolder implements
ServerReferencesProvider
{
return jsonMapper;
}
+
+ @Override
+ public <T> T getInstance(Class<T> clazz)
+ {
+ return injector.getInstance(clazz);
+ }
}
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/ServerReferencesProvider.java
b/services/src/test/java/org/apache/druid/testing/embedded/ServerReferencesProvider.java
index 6eeed490e1e..ca24218e0fb 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/ServerReferencesProvider.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/ServerReferencesProvider.java
@@ -99,4 +99,10 @@ public interface ServerReferencesProvider
* {@link ObjectMapper} annotated with {@link Json}.
*/
ObjectMapper jsonMapper();
+
+ /**
+ * Gets the injected instance of the object of the specified type.
+ * The returned object must be used for read-only purposes.
+ */
+ <T> T getInstance(Class<T> clazz);
}
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
b/services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
similarity index 100%
rename from
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
rename to
services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
index f103e7dbe70..76e6f37ab59 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
@@ -20,8 +20,6 @@
package org.apache.druid.sql.calcite.schema;
import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
@@ -29,7 +27,6 @@ import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.Futures;
import com.google.inject.Inject;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.ints.IntSet;
@@ -49,24 +46,18 @@ import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.druid.client.DruidServer;
import org.apache.druid.client.FilteredServerInventoryView;
import org.apache.druid.client.ImmutableDruidServer;
-import org.apache.druid.client.JsonParserIterator;
import org.apache.druid.client.TimelineServerView;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.discovery.DataNodeService;
import org.apache.druid.discovery.DiscoveryDruidNode;
-import org.apache.druid.discovery.DruidLeaderClient;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.discovery.NodeRole;
import org.apache.druid.indexer.TaskStatusPlus;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.common.RE;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.parsers.CloseableIterator;
-import org.apache.druid.java.util.http.client.Request;
-import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHandler;
-import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHolder;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
@@ -86,10 +77,8 @@ import org.apache.druid.sql.calcite.table.RowSignatures;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentStatusInCluster;
-import org.jboss.netty.handler.codec.http.HttpMethod;
import javax.annotation.Nullable;
-import javax.servlet.http.HttpServletResponse;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
@@ -1057,50 +1046,6 @@ public class SystemSchema extends AbstractSchema
}
}
- public static <T> JsonParserIterator<T> getThingsFromLeaderNode(
- String query,
- TypeReference<T> typeRef,
- DruidLeaderClient leaderClient,
- ObjectMapper jsonMapper
- )
- {
- Request request;
- InputStreamFullResponseHolder responseHolder;
- try {
- request = leaderClient.makeRequest(
- HttpMethod.GET,
- query
- );
-
- responseHolder = leaderClient.go(
- request,
- new InputStreamFullResponseHandler()
- );
-
- if (responseHolder.getStatus().getCode() != HttpServletResponse.SC_OK) {
- throw new RE(
- "Failed to talk to leader node at [%s]. Error code [%d],
description [%s].",
- query,
- responseHolder.getStatus().getCode(),
- responseHolder.getStatus().getReasonPhrase()
- );
- }
- }
- catch (IOException | InterruptedException e) {
- throw new RuntimeException(e);
- }
-
- final JavaType javaType =
jsonMapper.getTypeFactory().constructType(typeRef);
- return new JsonParserIterator<>(
- javaType,
- Futures.immediateFuture(responseHolder.getContent()),
- request.getUrl().toString(),
- null,
- request.getUrl().getHost(),
- jsonMapper
- );
- }
-
private static <T> CloseableIterator<T> wrap(Iterator<T> iterator, Closeable
closer)
{
return new CloseableIterator<>()
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/DruidCalciteSchemaModuleTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/DruidCalciteSchemaModuleTest.java
index 6b4153de28b..23335615409 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/DruidCalciteSchemaModuleTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/DruidCalciteSchemaModuleTest.java
@@ -30,11 +30,8 @@ import com.google.inject.TypeLiteral;
import com.google.inject.name.Names;
import org.apache.druid.client.FilteredServerInventoryView;
import org.apache.druid.client.TimelineServerView;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.client.coordinator.NoopCoordinatorClient;
-import org.apache.druid.client.indexing.IndexingService;
-import org.apache.druid.discovery.DruidLeaderClient;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.guice.LazySingleton;
import org.apache.druid.guice.LifecycleModule;
@@ -87,10 +84,6 @@ public class DruidCalciteSchemaModuleTest extends
CalciteTestBase
@Mock
private FilteredServerInventoryView serverInventoryView;
@Mock
- private DruidLeaderClient coordinatorDruidLeaderClient;
- @Mock
- private DruidLeaderClient overlordDruidLeaderClient;
- @Mock
private DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
@Mock
private ObjectMapper objectMapper;
@@ -121,12 +114,6 @@ public class DruidCalciteSchemaModuleTest extends
CalciteTestBase
binder.bind(FilteredServerInventoryView.class).toInstance(serverInventoryView);
binder.bind(SegmentManager.class).toInstance(segmentManager);
binder.bind(DruidOperatorTable.class).toInstance(druidOperatorTable);
- binder.bind(DruidLeaderClient.class)
- .annotatedWith(Coordinator.class)
- .toInstance(coordinatorDruidLeaderClient);
- binder.bind(DruidLeaderClient.class)
- .annotatedWith(IndexingService.class)
- .toInstance(overlordDruidLeaderClient);
binder.bind(DruidNodeDiscoveryProvider.class).toInstance(druidNodeDiscoveryProvider);
binder.bind(DruidSchemaManager.class).toInstance(new
NoopDruidSchemaManager());
binder.bind(ObjectMapper.class).annotatedWith(Json.class).toInstance(objectMapper);
diff --git a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
index 2dbb1a0bee1..08c7a0ea8cc 100644
--- a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
@@ -29,11 +29,8 @@ import com.google.inject.Key;
import com.google.inject.TypeLiteral;
import org.apache.druid.client.FilteredServerInventoryView;
import org.apache.druid.client.TimelineServerView;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.client.coordinator.NoopCoordinatorClient;
-import org.apache.druid.client.indexing.IndexingService;
-import org.apache.druid.discovery.DruidLeaderClient;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.guice.DruidGuiceExtensions;
import org.apache.druid.guice.JsonConfigurator;
@@ -98,9 +95,6 @@ public class SqlModuleTest
@Mock
private TimelineServerView timelineServerView;
- @Mock
- private DruidLeaderClient druidLeaderClient;
-
@Mock
private DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
@@ -134,7 +128,6 @@ public class SqlModuleTest
serviceEmitter,
inventoryView,
timelineServerView,
- druidLeaderClient,
druidNodeDiscoveryProvider,
genericQueryMetricsFactory,
querySegmentWalker,
@@ -201,8 +194,6 @@ public class SqlModuleTest
binder.bind(new
TypeLiteral<Supplier<DefaultQueryConfig>>(){}).toInstance(Suppliers.ofInstance(new
DefaultQueryConfig(null)));
binder.bind(FilteredServerInventoryView.class).toInstance(inventoryView);
binder.bind(TimelineServerView.class).toInstance(timelineServerView);
-
binder.bind(DruidLeaderClient.class).annotatedWith(Coordinator.class).toInstance(druidLeaderClient);
-
binder.bind(DruidLeaderClient.class).annotatedWith(IndexingService.class).toInstance(druidLeaderClient);
binder.bind(DruidNodeDiscoveryProvider.class).toInstance(druidNodeDiscoveryProvider);
binder.bind(GenericQueryMetricsFactory.class).toInstance(genericQueryMetricsFactory);
binder.bind(QuerySegmentWalker.class).toInstance(querySegmentWalker);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
