commit

apache · Nov 3, 2024 · 1377975 · 1377975
1 parent 9988d8f
commit 1377975
Show file tree

Hide file tree

Showing 19 changed files with 483 additions and 110 deletions.
diff --git a/...e/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java b/...e/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java
@@ -28,6 +28,7 @@
 import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskIOConfig;
 import org.apache.druid.indexing.seekablestream.SeekableStreamStartSequenceNumbers;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 
 import javax.annotation.Nullable;
 import java.util.Map;
@@ -53,7 +54,9 @@ public RabbitStreamIndexTaskIOConfig(
       @JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
       @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
       @JsonProperty("inputFormat") @Nullable InputFormat inputFormat,
-      @JsonProperty("uri") String uri)
+      @JsonProperty("uri") String uri,
+      @JsonProperty("taskDuration") Duration taskDuration
+  )
   {
     super(
         taskGroupId,
@@ -63,7 +66,9 @@ public RabbitStreamIndexTaskIOConfig(
         useTransaction,
         minimumMessageTime,
         maximumMessageTime,
-        inputFormat);
+        inputFormat,
+        taskDuration
+    );
 
     this.pollTimeout = pollTimeout != null ? pollTimeout : RabbitStreamSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS;
     this.uri = uri;

diff --git a/...c/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java b/...c/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java
@@ -59,7 +59,6 @@
 import org.joda.time.DateTime;
 
 import javax.annotation.Nullable;
-
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -202,7 +201,9 @@ protected SeekableStreamIndexTaskIOConfig createTaskIoConfig(
         minimumMessageTime,
         maximumMessageTime,
         ioConfig.getInputFormat(),
-        rabbitConfig.getUri());
+        rabbitConfig.getUri(),
+        ioConfig.getTaskDuration()
+    );
   }
 
   @Override

diff --git a/...ndexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskIOConfig.java b/...ndexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskIOConfig.java
@@ -30,6 +30,7 @@
 import org.apache.druid.indexing.seekablestream.SeekableStreamStartSequenceNumbers;
 import org.apache.druid.indexing.seekablestream.extension.KafkaConfigOverrides;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 
 import javax.annotation.Nullable;
 import java.util.Map;
@@ -63,7 +64,8 @@ public KafkaIndexTaskIOConfig(
       @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
       @JsonProperty("inputFormat") @Nullable InputFormat inputFormat,
       @JsonProperty("configOverrides") @Nullable KafkaConfigOverrides configOverrides,
-      @JsonProperty("multiTopic") @Nullable Boolean multiTopic
+      @JsonProperty("multiTopic") @Nullable Boolean multiTopic,
+      @JsonProperty("taskDuration") Duration taskDuration
   )
   {
     super(
@@ -76,7 +78,8 @@ public KafkaIndexTaskIOConfig(
         useTransaction,
         minimumMessageTime,
         maximumMessageTime,
-        inputFormat
+        inputFormat,
+        taskDuration
     );
 
     this.consumerProperties = Preconditions.checkNotNull(consumerProperties, "consumerProperties");
@@ -107,7 +110,8 @@ public KafkaIndexTaskIOConfig(
       DateTime minimumMessageTime,
       DateTime maximumMessageTime,
       InputFormat inputFormat,
-      KafkaConfigOverrides configOverrides
+      KafkaConfigOverrides configOverrides,
+      Duration taskDuration
   )
   {
     this(
@@ -124,7 +128,8 @@ public KafkaIndexTaskIOConfig(
         maximumMessageTime,
         inputFormat,
         configOverrides,
-        KafkaSupervisorIOConfig.DEFAULT_IS_MULTI_TOPIC
+        KafkaSupervisorIOConfig.DEFAULT_IS_MULTI_TOPIC,
+        taskDuration
     );
   }
 

diff --git a/...ing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java b/...ing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
@@ -219,7 +219,8 @@ protected SeekableStreamIndexTaskIOConfig createTaskIoConfig(
         maximumMessageTime,
         ioConfig.getInputFormat(),
         kafkaIoConfig.getConfigOverrides(),
-        kafkaIoConfig.isMultiTopic()
+        kafkaIoConfig.isMultiTopic(),
+        ioConfig.getTaskDuration()
     );
   }