You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by jo...@apache.org on 2019/12/21 03:35:19 UTC
[druid] branch master updated: Get proper Kinesis index task AWS
credentials (#9082)
This is an automated email from the ASF dual-hosted git repository.
jonwei 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 513bb1f Get proper Kinesis index task AWS credentials (#9082)
513bb1f is described below
commit 513bb1f6da56d6fe3120d376669309e6d1aaff67
Author: Chi Cao Minh <ch...@imply.io>
AuthorDate: Fri Dec 20 19:35:05 2019 -0800
Get proper Kinesis index task AWS credentials (#9082)
Previously, the configured S3 credentials would be used instead of the
ones configured for Kinesis for Kinesis index tasks.
---
extensions-core/kinesis-indexing-service/pom.xml | 5 +
.../druid/indexing/kinesis/KinesisIndexTask.java | 10 +-
.../kinesis/KinesisIndexingServiceModule.java | 5 +-
.../druid/indexing/kinesis/KinesisSamplerSpec.java | 2 +-
.../kinesis/supervisor/KinesisSupervisorSpec.java | 3 +-
.../kinesis/KinesisIndexTaskSerdeTest.java | 156 +++++++++++++++++++++
.../indexing/kinesis/KinesisIndexTaskTest.java | 3 +-
7 files changed, 179 insertions(+), 5 deletions(-)
diff --git a/extensions-core/kinesis-indexing-service/pom.xml b/extensions-core/kinesis-indexing-service/pom.xml
index 590aa94..ea4afae 100644
--- a/extensions-core/kinesis-indexing-service/pom.xml
+++ b/extensions-core/kinesis-indexing-service/pom.xml
@@ -188,6 +188,11 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.github.stefanbirkner</groupId>
+ <artifactId>system-rules</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
index e8f49bd..5d5a307 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
@@ -22,6 +22,8 @@ package org.apache.druid.indexing.kinesis;
import com.fasterxml.jackson.annotation.JacksonInject;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.name.Named;
import org.apache.druid.common.aws.AWSCredentialsConfig;
import org.apache.druid.indexing.common.stats.RowIngestionMetersFactory;
import org.apache.druid.indexing.common.task.TaskResource;
@@ -51,7 +53,7 @@ public class KinesisIndexTask extends SeekableStreamIndexTask<String, String>
@JacksonInject ChatHandlerProvider chatHandlerProvider,
@JacksonInject AuthorizerMapper authorizerMapper,
@JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
- @JacksonInject AWSCredentialsConfig awsCredentialsConfig,
+ @JacksonInject @Named(KinesisIndexingServiceModule.AWS_SCOPE) AWSCredentialsConfig awsCredentialsConfig,
@JacksonInject AppenderatorsManager appenderatorsManager
)
{
@@ -128,4 +130,10 @@ public class KinesisIndexTask extends SeekableStreamIndexTask<String, String>
{
return TYPE;
}
+
+ @VisibleForTesting
+ AWSCredentialsConfig getAwsCredentialsConfig()
+ {
+ return awsCredentialsConfig;
+ }
}
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexingServiceModule.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexingServiceModule.java
index 3cea45a..bd11e4a 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexingServiceModule.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexingServiceModule.java
@@ -38,6 +38,9 @@ import java.util.List;
public class KinesisIndexingServiceModule implements DruidModule
{
+ public static final String AWS_SCOPE = "kinesis";
+ static final String PROPERTY_BASE = "druid.kinesis";
+
@Override
public List<? extends Module> getJacksonModules()
{
@@ -63,6 +66,6 @@ public class KinesisIndexingServiceModule implements DruidModule
}
).to(KinesisIndexTaskClientFactory.class).in(LazySingleton.class);
- JsonConfigProvider.bind(binder, "druid.kinesis", AWSCredentialsConfig.class, Names.named("kinesis"));
+ JsonConfigProvider.bind(binder, PROPERTY_BASE, AWSCredentialsConfig.class, Names.named(AWS_SCOPE));
}
}
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpec.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpec.java
index dbe77a2..664b3b5 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpec.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpec.java
@@ -42,7 +42,7 @@ public class KinesisSamplerSpec extends SeekableStreamSamplerSpec
@JsonProperty("spec") final KinesisSupervisorSpec ingestionSpec,
@JsonProperty("samplerConfig") @Nullable final SamplerConfig samplerConfig,
@JacksonInject InputSourceSampler inputSourceSampler,
- @JacksonInject @Named("kinesis") AWSCredentialsConfig awsCredentialsConfig
+ @JacksonInject @Named(KinesisIndexingServiceModule.AWS_SCOPE) AWSCredentialsConfig awsCredentialsConfig
)
{
super(ingestionSpec, samplerConfig, inputSourceSampler);
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
index e9610c6..3092d3a 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
@@ -28,6 +28,7 @@ import org.apache.druid.common.aws.AWSCredentialsConfig;
import org.apache.druid.guice.annotations.Json;
import org.apache.druid.indexing.common.stats.RowIngestionMetersFactory;
import org.apache.druid.indexing.kinesis.KinesisIndexTaskClientFactory;
+import org.apache.druid.indexing.kinesis.KinesisIndexingServiceModule;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskStorage;
@@ -62,7 +63,7 @@ public class KinesisSupervisorSpec extends SeekableStreamSupervisorSpec
@JacksonInject ServiceEmitter emitter,
@JacksonInject DruidMonitorSchedulerConfig monitorSchedulerConfig,
@JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
- @JacksonInject @Named("kinesis") AWSCredentialsConfig awsCredentialsConfig,
+ @JacksonInject @Named(KinesisIndexingServiceModule.AWS_SCOPE) AWSCredentialsConfig awsCredentialsConfig,
@JacksonInject SupervisorStateManagerConfig supervisorStateManagerConfig
)
{
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskSerdeTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskSerdeTest.java
new file mode 100644
index 0000000..646a334
--- /dev/null
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskSerdeTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.indexing.kinesis;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.name.Names;
+import org.apache.druid.common.aws.AWSCredentialsConfig;
+import org.apache.druid.guice.GuiceInjectors;
+import org.apache.druid.indexing.common.stats.DropwizardRowIngestionMetersFactory;
+import org.apache.druid.indexing.common.stats.RowIngestionMetersFactory;
+import org.apache.druid.indexing.common.task.TestAppenderatorsManager;
+import org.apache.druid.indexing.seekablestream.SeekableStreamEndSequenceNumbers;
+import org.apache.druid.indexing.seekablestream.SeekableStreamStartSequenceNumbers;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.initialization.Initialization;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
+import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
+import org.apache.druid.segment.realtime.firehose.NoopChatHandlerProvider;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.ProvideSystemProperty;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public class KinesisIndexTaskSerdeTest
+{
+ private static final DataSchema DATA_SCHEMA = new DataSchema("dataSource", null, null, null, null, null, null, null);
+ private static final KinesisIndexTaskTuningConfig TUNING_CONFIG = new KinesisIndexTaskTuningConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+ private static final KinesisIndexTaskIOConfig IO_CONFIG = new KinesisIndexTaskIOConfig(
+ 0,
+ "baseSequenceName",
+ new SeekableStreamStartSequenceNumbers<>("stream", Collections.emptyMap(), null),
+ new SeekableStreamEndSequenceNumbers<>("stream", Collections.emptyMap()),
+ null,
+ null,
+ null,
+ null,
+ "endpoint",
+ null,
+ null,
+ null,
+ null,
+ false
+ );
+ private static final String ACCESS_KEY = "test-access-key";
+ private static final String SECRET_KEY = "test-secret-key";
+ private static final String FILE_SESSION_CREDENTIALS = "test-file-session-credentials";
+
+ @Rule
+ public ProvideSystemProperty properties = new ProvideSystemProperty(
+ KinesisIndexingServiceModule.PROPERTY_BASE + ".accessKey",
+ ACCESS_KEY
+ ).and(
+ KinesisIndexingServiceModule.PROPERTY_BASE + ".secretKey",
+ SECRET_KEY
+ ).and(
+ KinesisIndexingServiceModule.PROPERTY_BASE + ".fileSessionCredentials",
+ FILE_SESSION_CREDENTIALS
+ );
+
+ @Test
+ public void injectsProperAwsCredentialsConfig() throws Exception
+ {
+ KinesisIndexTask target = new KinesisIndexTask(
+ "id",
+ null,
+ DATA_SCHEMA,
+ TUNING_CONFIG,
+ IO_CONFIG,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+ ObjectMapper objectMapper = createObjectMapper();
+ String serialized = objectMapper.writeValueAsString(target);
+ KinesisIndexTask deserialized = objectMapper.readValue(serialized, KinesisIndexTask.class);
+
+ AWSCredentialsConfig awsCredentialsConfig = deserialized.getAwsCredentialsConfig();
+ Assert.assertEquals(ACCESS_KEY, awsCredentialsConfig.getAccessKey().getPassword());
+ Assert.assertEquals(SECRET_KEY, awsCredentialsConfig.getSecretKey().getPassword());
+ Assert.assertEquals(FILE_SESSION_CREDENTIALS, awsCredentialsConfig.getFileSessionCredentials());
+ }
+
+ private static ObjectMapper createObjectMapper()
+ {
+ DruidModule module = new KinesisIndexingServiceModule();
+ Injector injector = Initialization.makeInjectorWithModules(
+ GuiceInjectors.makeStartupInjector(),
+ Arrays.asList(
+ module,
+ (Module) binder -> {
+ binder.bindConstant().annotatedWith(Names.named("serviceName")).to("test");
+ binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8000);
+ binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9000);
+ binder.bind(ChatHandlerProvider.class).toInstance(new NoopChatHandlerProvider());
+ binder.bind(RowIngestionMetersFactory.class).toInstance(new DropwizardRowIngestionMetersFactory());
+ binder.bind(AppenderatorsManager.class).toInstance(new TestAppenderatorsManager());
+ }
+ )
+ );
+ ObjectMapper objectMapper = injector.getInstance(ObjectMapper.class);
+ module.getJacksonModules().forEach(objectMapper::registerModule);
+ return objectMapper;
+ }
+}
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
index a334a79..c214def 100644
--- a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
@@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.inject.name.Named;
import org.apache.druid.client.cache.CacheConfig;
import org.apache.druid.client.cache.CachePopulatorStats;
import org.apache.druid.client.cache.MapCache;
@@ -2939,7 +2940,7 @@ public class KinesisIndexTaskTest extends SeekableStreamIndexTaskTestBase
@JacksonInject ChatHandlerProvider chatHandlerProvider,
@JacksonInject AuthorizerMapper authorizerMapper,
@JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
- @JacksonInject AWSCredentialsConfig awsCredentialsConfig,
+ @JacksonInject @Named(KinesisIndexingServiceModule.AWS_SCOPE) AWSCredentialsConfig awsCredentialsConfig,
@JacksonInject AppenderatorsManager appenderatorsManager
)
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org