add context to kafka supervisor for the kafka indexing task (apache#3464

)
fundead · Dec 7, 2016 · 80b0e4e · 80b0e4e
1 parent 8d4394e
commit 80b0e4e
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 1 deletion.
diff --git a/...ka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java b/...ka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java
@@ -1272,7 +1272,7 @@ private void createKafkaTasksForGroup(int groupId, int replicas)
           spec.getDataSchema(),
           taskTuningConfig,
           kafkaIOConfig,
-          ImmutableMap.<String, Object>of(),
+          spec.getContext(),
           null
       );
 

diff --git a/...ndexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java b/...ndexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
@@ -33,11 +33,14 @@
 import io.druid.indexing.overlord.supervisor.SupervisorSpec;
 import io.druid.segment.indexing.DataSchema;
 
+import java.util.Map;
+
 public class KafkaSupervisorSpec implements SupervisorSpec
 {
   private final DataSchema dataSchema;
   private final KafkaSupervisorTuningConfig tuningConfig;
   private final KafkaSupervisorIOConfig ioConfig;
+  private final Map<String, Object> context;
 
   private final TaskStorage taskStorage;
   private final TaskMaster taskMaster;
@@ -50,6 +53,7 @@ public KafkaSupervisorSpec(
       @JsonProperty("dataSchema") DataSchema dataSchema,
       @JsonProperty("tuningConfig") KafkaSupervisorTuningConfig tuningConfig,
       @JsonProperty("ioConfig") KafkaSupervisorIOConfig ioConfig,
+      @JsonProperty("context") Map<String, Object> context,
       @JacksonInject TaskStorage taskStorage,
       @JacksonInject TaskMaster taskMaster,
       @JacksonInject IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator,
@@ -77,6 +81,7 @@ public KafkaSupervisorSpec(
                             null
                         );
     this.ioConfig = Preconditions.checkNotNull(ioConfig, "ioConfig");
+    this.context = context;
 
     this.taskStorage = taskStorage;
     this.taskMaster = taskMaster;
@@ -103,6 +108,12 @@ public KafkaSupervisorIOConfig getIoConfig()
     return ioConfig;
   }
 
+  @JsonProperty
+  public Map<String, Object> getContext()
+  {
+    return context;
+  }
+
   @Override
   public String getId()
   {

diff --git a/...ndexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java b/...ndexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
@@ -1632,6 +1632,7 @@ public KafkaIndexTaskClient build(
             dataSchema,
             tuningConfig,
             kafkaSupervisorIOConfig,
+            null,
             taskStorage,
             taskMaster,
             indexerMetadataStorageCoordinator,