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 a352190b4d1 Remove `IS_INCREMENTAL_HANDOFF_SUPPORTED` legacy context
reference from KafkaSupervisor (#18120)
a352190b4d1 is described below
commit a352190b4d1b1b507e8a38f858ea27c9a887202f
Author: Abhishek Radhakrishnan <[email protected]>
AuthorDate: Wed Jun 11 19:31:23 2025 -0700
Remove `IS_INCREMENTAL_HANDOFF_SUPPORTED` legacy context reference from
KafkaSupervisor (#18120)
Removed the `IS_INCREMENTAL_HANDOFF_SUPPORTED` context reference from
supervisors,
as incremental publishing has been the default behavior since version
0.16.0. This context was
originally introduced to support rollback to `LegacyKafkaIndexTaskRunner`
in versions earlier
than 0.16.0, which has since been removed.
---
.../druid/indexing/rabbitstream/RabbitStreamIndexTask.java | 2 +-
...IndexTaskRunner.java => RabbitStreamIndexTaskRunner.java} | 8 ++++----
.../java/org/apache/druid/indexing/kafka/KafkaIndexTask.java | 2 +-
...ngKafkaIndexTaskRunner.java => KafkaIndexTaskRunner.java} | 8 ++++----
.../druid/indexing/kafka/supervisor/KafkaSupervisor.java | 4 ----
.../druid/indexing/kafka/supervisor/KafkaSupervisorTest.java | 12 ++++++------
6 files changed, 16 insertions(+), 20 deletions(-)
diff --git
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTask.java
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTask.java
index 4fb452e0b2b..416df49796d 100644
---
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTask.java
+++
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTask.java
@@ -86,7 +86,7 @@ public class RabbitStreamIndexTask extends
SeekableStreamIndexTask<String, Long,
protected SeekableStreamIndexTaskRunner<String, Long, ByteEntity>
createTaskRunner()
{
// noinspection unchecked
- return new IncrementalPublishingRabbitStreamIndexTaskRunner(
+ return new RabbitStreamIndexTaskRunner(
this,
dataSchema.getParser(),
lockGranularityToUse
diff --git
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/IncrementalPublishingRabbitStreamIndexTaskRunner.java
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskRunner.java
similarity index 93%
rename from
extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/IncrementalPublishingRabbitStreamIndexTaskRunner.java
rename to
extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskRunner.java
index 51f9fd901a0..bfe93dedbfb 100644
---
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/IncrementalPublishingRabbitStreamIndexTaskRunner.java
+++
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskRunner.java
@@ -47,15 +47,15 @@ import java.util.Set;
import java.util.TreeMap;
/**
- * RabbitStream indexing task runner supporting incremental segments publishing
+ * RabbitStream indexing task runner that supports incremental segment
publishing.
*/
-public class IncrementalPublishingRabbitStreamIndexTaskRunner
+public class RabbitStreamIndexTaskRunner
extends SeekableStreamIndexTaskRunner<String, Long, ByteEntity>
{
- private static final EmittingLogger log = new
EmittingLogger(IncrementalPublishingRabbitStreamIndexTaskRunner.class);
+ private static final EmittingLogger log = new
EmittingLogger(RabbitStreamIndexTaskRunner.class);
private final RabbitStreamIndexTask task;
- IncrementalPublishingRabbitStreamIndexTaskRunner(
+ RabbitStreamIndexTaskRunner(
RabbitStreamIndexTask task,
@Nullable InputRowParser<ByteBuffer> parser,
LockGranularity lockGranularityToUse
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
index 3f64a6d05e5..b4846d728e2 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
@@ -95,7 +95,7 @@ public class KafkaIndexTask extends
SeekableStreamIndexTask<KafkaTopicPartition,
protected SeekableStreamIndexTaskRunner<KafkaTopicPartition, Long,
KafkaRecordEntity> createTaskRunner()
{
//noinspection unchecked
- return new IncrementalPublishingKafkaIndexTaskRunner(
+ return new KafkaIndexTaskRunner(
this,
dataSchema.getParser(),
lockGranularityToUse
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
similarity index 95%
rename from
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
rename to
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
index 73270c18576..8dbcbdcf92c 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
@@ -55,14 +55,14 @@ import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
/**
- * Kafka indexing task runner supporting incremental segments publishing
+ * Kafka indexing task runner that supports incremental segment publishing.
*/
-public class IncrementalPublishingKafkaIndexTaskRunner extends
SeekableStreamIndexTaskRunner<KafkaTopicPartition, Long, KafkaRecordEntity>
+public class KafkaIndexTaskRunner extends
SeekableStreamIndexTaskRunner<KafkaTopicPartition, Long, KafkaRecordEntity>
{
- private static final EmittingLogger log = new
EmittingLogger(IncrementalPublishingKafkaIndexTaskRunner.class);
+ private static final EmittingLogger log = new
EmittingLogger(KafkaIndexTaskRunner.class);
private final KafkaIndexTask task;
- IncrementalPublishingKafkaIndexTaskRunner(
+ KafkaIndexTaskRunner(
KafkaIndexTask task,
@Nullable InputRowParser<ByteBuffer> parser,
LockGranularity lockGranularityToUse
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
index 743e4edc32c..82905b8eb92 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
@@ -231,10 +231,6 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
final String checkpoints =
sortingMapper.writerFor(CHECKPOINTS_TYPE_REF).writeValueAsString(sequenceOffsets);
final Map<String, Object> context = createBaseTaskContexts();
context.put(CHECKPOINTS_CTX_KEY, checkpoints);
- // Kafka index task always uses incremental handoff since 0.16.0.
- // The below is for the compatibility when you want to downgrade your
cluster to something earlier than 0.16.0.
- // Kafka index task will pick up LegacyKafkaIndexTaskRunner without the
below configuration.
- context.put("IS_INCREMENTAL_HANDOFF_SUPPORTED", true);
List<SeekableStreamIndexTask<KafkaTopicPartition, Long,
KafkaRecordEntity>> taskList = new ArrayList<>();
for (int i = 0; i < replicas; i++) {
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
index 9428de463bf..814a9bb5080 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
@@ -47,6 +47,7 @@ import
org.apache.druid.indexing.kafka.KafkaDataSourceMetadata;
import org.apache.druid.indexing.kafka.KafkaIndexTask;
import org.apache.druid.indexing.kafka.KafkaIndexTaskClientFactory;
import org.apache.druid.indexing.kafka.KafkaIndexTaskIOConfig;
+import org.apache.druid.indexing.kafka.KafkaIndexTaskRunner;
import org.apache.druid.indexing.kafka.KafkaIndexTaskTuningConfig;
import org.apache.druid.indexing.kafka.KafkaRecordSupplier;
import org.apache.druid.indexing.kafka.test.TestBroker;
@@ -63,6 +64,7 @@ import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManager;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
import
org.apache.druid.indexing.seekablestream.SeekableStreamEndSequenceNumbers;
+import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask;
import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskClient;
import
org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.Status;
import
org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskTuningConfig;
@@ -485,10 +487,10 @@ public class KafkaSupervisorTest extends EasyMockSupport
}
@Test
- public void testCreateBaseTaskContexts() throws JsonProcessingException
+ public void testGetTaskRunnerType() throws JsonProcessingException
{
supervisor = getTestableSupervisor(1, 1, true, "PT1H", null, null);
- final Map<String, Object> contexts = supervisor.createIndexTasks(
+ final SeekableStreamIndexTask<KafkaTopicPartition, Long,
KafkaRecordEntity> indexTask = supervisor.createIndexTasks(
1,
"seq",
OBJECT_MAPPER,
@@ -531,10 +533,8 @@ public class KafkaSupervisorTest extends EasyMockSupport
null
),
null
- ).get(0).getContext();
- final Boolean contextValue = (Boolean)
contexts.get("IS_INCREMENTAL_HANDOFF_SUPPORTED");
- Assert.assertNotNull(contextValue);
- Assert.assertTrue(contextValue);
+ ).get(0);
+ Assert.assertTrue(indexTask.getRunner() instanceof KafkaIndexTaskRunner);
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
