You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "aho135 (via GitHub)" <gi...@apache.org> on 2026/04/24 17:30:07 UTC
[PR] Bounded stream supervisor (druid)
aho135 opened a new pull request, #19372:
URL: https://github.com/apache/druid/pull/19372
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
Introduces a new property to the Stream Supervisor spec IOConfig called boundedStreamConfig which allows operators to specify start and end offset ranges for short-lived supervised ingestion. This property modifies the main Supervisor run loop to only ingest from a monitor partitions specified in the boundedStreamConfig. After the offset range has been consumed the Supervisor will transition into a terminal state (COMPLETED). The motivation for this PR came out of https://github.com/apache/druid/pull/19191 which submits backfill tasks that are unsupervised. One this change is merged, 19191 can be enhanced to use the boundedStreamConfig so that the backfill tasks are supervised.
<!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
<!-- Describe your patch: what did you change in code? How did you fix the problem? -->
<!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
Adds a property called boundedStreamConfig to the SeekableStreamSupervisorIOConfig which allows operators to spin up a Supervisor that consumes only a specified offset range.
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `SeekableStreamSupervisorIOConfig`
* `BoundedStreamConfig`
* `SeekableStreamSupervisor`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [X] been self-reviewed.
- [X] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [X] a release note entry in the PR description.
- [X] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [X] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [X] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [X] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3140866724
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java:
##########
@@ -78,7 +79,8 @@ public KafkaSupervisorIOConfig(
@JsonProperty("idleConfig") IdleConfig idleConfig,
@JsonProperty("stopTaskCount") Integer stopTaskCount,
@Nullable @JsonProperty("emitTimeLagMetrics") Boolean emitTimeLagMetrics,
- @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas
+ @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas,
+ @Nullable @JsonProperty("boundedStreamConfig") BoundedStreamConfig boundedStreamConfig
Review Comment:
It would also be good to update some user-facing documentation in `supervisor.md` for this functionality
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3165233350
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
+ *
+ * When configured, the supervisor will:
+ * 1. Create tasks starting at the specified startSequenceNumbers
+ * 2. Tasks will automatically stop when they reach endSequenceNumbers
+ * 3. Supervisor will not recreate tasks after they complete
+ * 4. Supervisor will auto-terminate when all tasks are done
+ *
+ * This is useful for:
+ * - Backfill processing
+ * - Historical reprocessing
+ * - One-time migration tasks
+ */
+public class BoundedStreamConfig
+{
+ private final Map<?, ?> startSequenceNumbers; // Partition -> Start Offset
+ private final Map<?, ?> endSequenceNumbers; // Partition -> End Offset
+
+ @JsonCreator
+ public BoundedStreamConfig(
+ @JsonProperty("startSequenceNumbers") Map<?, ?> startSequenceNumbers,
+ @JsonProperty("endSequenceNumbers") Map<?, ?> endSequenceNumbers
+ )
+ {
+ this.startSequenceNumbers = Preconditions.checkNotNull(startSequenceNumbers, "startSequenceNumbers");
+ this.endSequenceNumbers = Preconditions.checkNotNull(endSequenceNumbers, "endSequenceNumbers");
+
+ // Validation
Review Comment:
This validation is a bit difficult to do within BoundedStreamConfig because of the generic typing. But I do see there is already validation for this in Task IOConfig. [kafka](https://github.com/apache/druid/blob/master/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskIOConfig.java#L91-L97)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3223773301
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * Regression test: a new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED. Before the fix, hasTaskGroupReachedBoundedEnd() compared
+ * the stale committed offset against the new endOffset (e.g. committed=100 >= newEnd=50) and
+ * returned true, bypassing task creation and the documented mismatch error entirely.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
+
+ cluster.callApi().postSupervisor(supervisor1);
+ waitUntilPublishedRecordsAreIngested(200);
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Run 2: same datasource, endOffset (50) < stale committed offset (100).
+ // Without the fix the supervisor reaches COMPLETED immediately without running tasks.
+ // With the fix it detects the config mismatch and becomes UNHEALTHY_SUPERVISOR.
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 0L);
+ startOffsets2.put("1", 0L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 50L);
+ endOffsets2.put("1", 50L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig2);
+
+ cluster.callApi().postSupervisor(supervisor2);
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
Review Comment:
Do you think validating a third supervisor with a different ID but the same bounds would be worth adding that succeeds? (either in this test or the following one)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3140866724
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java:
##########
@@ -78,7 +79,8 @@ public KafkaSupervisorIOConfig(
@JsonProperty("idleConfig") IdleConfig idleConfig,
@JsonProperty("stopTaskCount") Integer stopTaskCount,
@Nullable @JsonProperty("emitTimeLagMetrics") Boolean emitTimeLagMetrics,
- @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas
+ @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas,
+ @Nullable @JsonProperty("boundedStreamConfig") BoundedStreamConfig boundedStreamConfig
Review Comment:
I think it would also be good to update some user-facing documentation in `supervisor.md` for this functionality
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3165331522
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
+ *
+ * When configured, the supervisor will:
+ * 1. Create tasks starting at the specified startSequenceNumbers
+ * 2. Tasks will automatically stop when they reach endSequenceNumbers
+ * 3. Supervisor will not recreate tasks after they complete
+ * 4. Supervisor will auto-terminate when all tasks are done
+ *
+ * This is useful for:
+ * - Backfill processing
+ * - Historical reprocessing
+ * - One-time migration tasks
+ */
+public class BoundedStreamConfig
+{
+ private final Map<?, ?> startSequenceNumbers; // Partition -> Start Offset
+ private final Map<?, ?> endSequenceNumbers; // Partition -> End Offset
+
+ @JsonCreator
+ public BoundedStreamConfig(
+ @JsonProperty("startSequenceNumbers") Map<?, ?> startSequenceNumbers,
+ @JsonProperty("endSequenceNumbers") Map<?, ?> endSequenceNumbers
+ )
+ {
+ this.startSequenceNumbers = Preconditions.checkNotNull(startSequenceNumbers, "startSequenceNumbers");
+ this.endSequenceNumbers = Preconditions.checkNotNull(endSequenceNumbers, "endSequenceNumbers");
+
+ // Validation
Review Comment:
The validation there isn't strict enough though because startOffset can equal endOffset. In that scenario the Supervisor spins up a task that consumes nothing and then shuts down. But since no data was consumed there is no metadata update so it gets stuck in a loop where it keeps spinning up tasks. I added additional validation to handle this scenario in this [commit](https://github.com/apache/druid/pull/19372/changes/c9181f025713915977dc7bdc237069d160189791)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3139750629
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -230,6 +234,37 @@
groupId,
startingSequences,
unfilteredStartingSequencesForSequenceName,
+ null, // endSequences - null for streaming mode
+ minimumMessageTime,
+ maximumMessageTime,
+ exclusiveStartSequenceNumberPartitions,
+ generateSequenceName(
+ unfilteredStartingSequencesForSequenceName == null
+ ? startingSequences
Review Comment:
## CodeQL / Deprecated method or constructor invocation
Invoking [SeekableStreamSupervisorSpec.getDataSchema](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/560)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2609,6 +2614,24 @@
{
return new LagStats(0, 0, 0);
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
+ }
+
+ @Override
+ protected String createPartitionIdFromString(String partitionIdString)
+ {
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11133)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2609,6 +2614,24 @@
{
return new LagStats(0, 0, 0);
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
+ }
+
+ @Override
+ protected String createPartitionIdFromString(String partitionIdString)
+ {
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11134)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java:
##########
@@ -328,6 +328,24 @@
{
return partitionNumbers;
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11131)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -3273,6 +3299,24 @@
{
return false;
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
+ }
+
+ @Override
+ protected String createPartitionIdFromString(String partitionIdString)
+ {
+ return partitionIdString;
+ }
+
+ @Override
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11135)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java:
##########
@@ -328,6 +328,24 @@
{
return partitionNumbers;
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11132)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -3273,6 +3299,24 @@
{
return false;
}
+
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ return Long.parseLong(current) >= Long.parseLong(target);
+ }
+
+ @Override
+ protected String createPartitionIdFromString(String partitionIdString)
+ {
+ return partitionIdString;
+ }
+
+ @Override
Review Comment:
## CodeQL / Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/11136)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3140789053
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
+ *
+ * When configured, the supervisor will:
+ * 1. Create tasks starting at the specified startSequenceNumbers
+ * 2. Tasks will automatically stop when they reach endSequenceNumbers
+ * 3. Supervisor will not recreate tasks after they complete
+ * 4. Supervisor will auto-terminate when all tasks are done
+ *
+ * This is useful for:
+ * - Backfill processing
+ * - Historical reprocessing
+ * - One-time migration tasks
+ */
+public class BoundedStreamConfig
+{
+ private final Map<?, ?> startSequenceNumbers; // Partition -> Start Offset
+ private final Map<?, ?> endSequenceNumbers; // Partition -> End Offset
+
+ @JsonCreator
+ public BoundedStreamConfig(
+ @JsonProperty("startSequenceNumbers") Map<?, ?> startSequenceNumbers,
+ @JsonProperty("endSequenceNumbers") Map<?, ?> endSequenceNumbers
+ )
+ {
+ this.startSequenceNumbers = Preconditions.checkNotNull(startSequenceNumbers, "startSequenceNumbers");
+ this.endSequenceNumbers = Preconditions.checkNotNull(endSequenceNumbers, "endSequenceNumbers");
+
+ // Validation
Review Comment:
As a guard rail, I think it'll be good to have stricter checks for each partition so there's no unintended behavior with incorrect ranges specified:
`startSequenceNumbers < endSequenceNumbers`
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
Review Comment:
could you document that the start & end offsets are inclusive?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
+ *
+ * When configured, the supervisor will:
+ * 1. Create tasks starting at the specified startSequenceNumbers
+ * 2. Tasks will automatically stop when they reach endSequenceNumbers
+ * 3. Supervisor will not recreate tasks after they complete
+ * 4. Supervisor will auto-terminate when all tasks are done
+ *
+ * This is useful for:
+ * - Backfill processing
+ * - Historical reprocessing
+ * - One-time migration tasks
+ */
+public class BoundedStreamConfig
+{
+ private final Map<?, ?> startSequenceNumbers; // Partition -> Start Offset
+ private final Map<?, ?> endSequenceNumbers; // Partition -> End Offset
+
+ @JsonCreator
+ public BoundedStreamConfig(
+ @JsonProperty("startSequenceNumbers") Map<?, ?> startSequenceNumbers,
+ @JsonProperty("endSequenceNumbers") Map<?, ?> endSequenceNumbers
+ )
+ {
+ this.startSequenceNumbers = Preconditions.checkNotNull(startSequenceNumbers, "startSequenceNumbers");
+ this.endSequenceNumbers = Preconditions.checkNotNull(endSequenceNumbers, "endSequenceNumbers");
Review Comment:
nit: we could throw `DruidException`s here with the admin persona for these validation errors here
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorStateManager.java:
##########
@@ -64,7 +64,8 @@ public enum BasicState implements State
RUNNING(true, false),
IDLE(true, false),
SUSPENDED(true, false),
- STOPPING(true, false);
+ STOPPING(true, false),
+ COMPLETED(true, false);
Review Comment:
Just wondering, if it's possible to have a completed but unhealthy state? Or would that be `UNHEALTHY_SUPERVISOR`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3165138497
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
@FrankChen021 Thinking through this one a bit more, the validation approach is a bit tricky. It's not straightforward to tell if the existing metadata is from the bounded Supervisor itself (in which case starting from the metadata would be the correct behavior) or if it's from a previous Supervisor.
One approach we can take is that if the metadata offset falls within the configured start/end offsets then use that, otherwise fall back to startOffset. This does run the risk of partial ingestion of the specified range though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3175998966
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
Thanks @FrankChen021! I took a stab at this in the most recent [commit](https://github.com/apache/druid/pull/19372/changes/5965ac4de5f1e344906ddbe1a94afce29b85b04e)
Please let me know you thoughts when you get the chance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3140862918
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java:
##########
@@ -78,7 +79,8 @@ public KafkaSupervisorIOConfig(
@JsonProperty("idleConfig") IdleConfig idleConfig,
@JsonProperty("stopTaskCount") Integer stopTaskCount,
@Nullable @JsonProperty("emitTimeLagMetrics") Boolean emitTimeLagMetrics,
- @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas
+ @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas,
+ @Nullable @JsonProperty("boundedStreamConfig") BoundedStreamConfig boundedStreamConfig
Review Comment:
How do the bounds interplay with the supervisor’s behavior with other properties like `taskDuration`, `completionTimeout`, etc.?
I suppose if the bounds are too wide, tasks would still roll over and all existing properties would be honored as-is. The only additional behavioral change is when this optional config is specified and the end sequence numbers are reached, the supervisor eventually moves to a terminal `COMPLETED` state?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3223501358
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfigTest.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.error.DruidException;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class BoundedStreamConfigTest
+{
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ public void testConstructorWithValidMaps()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+ startOffsets.put("1", 200L);
+
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500L);
+ endOffsets.put("1", 600L);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ Assert.assertEquals(startOffsets, config.getStartSequenceNumbers());
+ Assert.assertEquals(endOffsets, config.getEndSequenceNumbers());
+ }
+
+ @Test
+ public void testConstructorWithNullStartSequenceNumbers()
+ {
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500L);
+
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new BoundedStreamConfig(null, endOffsets)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("cannot be null or empty"));
+ }
+
+ @Test
+ public void testConstructorWithNullEndSequenceNumbers()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new BoundedStreamConfig(startOffsets, null)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("cannot be null or empty"));
+ }
+
+ @Test
+ public void testConstructorWithEmptyStartSequenceNumbers()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500L);
+
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new BoundedStreamConfig(startOffsets, endOffsets)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("cannot be null or empty"));
+ }
+
+ @Test
+ public void testConstructorWithEmptyEndSequenceNumbers()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+ Map<String, Long> endOffsets = new HashMap<>();
+
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new BoundedStreamConfig(startOffsets, endOffsets)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("cannot be null or empty"));
+ }
+
+ @Test
+ public void testConstructorWithMismatchedPartitions()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("1", 500L);
+
+ DruidException ex = Assert.assertThrows(
+ DruidException.class,
+ () -> new BoundedStreamConfig(startOffsets, endOffsets)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("must have matching partition sets"));
+ }
+
+ @Test
+ public void testSerializationDeserialization() throws Exception
+ {
+ Map<String, Integer> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100);
+ startOffsets.put("1", 200);
+
+ Map<String, Integer> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500);
+ endOffsets.put("1", 600);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ String json = mapper.writeValueAsString(config);
+ BoundedStreamConfig deserialized = mapper.readValue(json, BoundedStreamConfig.class);
+
+ // Check sizes
+ Assert.assertEquals(2, deserialized.getStartSequenceNumbers().size());
+ Assert.assertEquals(2, deserialized.getEndSequenceNumbers().size());
+
+ // Check that deserialized maps contain expected values (keys will be Strings after deserialization)
+ Assert.assertEquals(100, deserialized.getStartSequenceNumbers().get("0"));
+ Assert.assertEquals(200, deserialized.getStartSequenceNumbers().get("1"));
+ Assert.assertEquals(500, deserialized.getEndSequenceNumbers().get("0"));
+ Assert.assertEquals(600, deserialized.getEndSequenceNumbers().get("1"));
+ }
+
+ @Test
+ public void testDeserializationWithIntegerValues() throws Exception
+ {
+ String json = "{"
+ + "\"startSequenceNumbers\": {\"0\": 100, \"1\": 200},"
+ + "\"endSequenceNumbers\": {\"0\": 500, \"1\": 600}"
+ + "}";
+
+ BoundedStreamConfig config = mapper.readValue(json, BoundedStreamConfig.class);
+
+ Assert.assertNotNull(config.getStartSequenceNumbers());
+ Assert.assertNotNull(config.getEndSequenceNumbers());
+ Assert.assertEquals(2, config.getStartSequenceNumbers().size());
+ Assert.assertEquals(2, config.getEndSequenceNumbers().size());
+ }
+
+ @Test
+ public void testDeserializationWithStringValues() throws Exception
+ {
+ String json = "{"
+ + "\"startSequenceNumbers\": {\"0\": \"100\", \"1\": \"200\"},"
+ + "\"endSequenceNumbers\": {\"0\": \"500\", \"1\": \"600\"}"
+ + "}";
+
+ BoundedStreamConfig config = mapper.readValue(json, BoundedStreamConfig.class);
+
+ Assert.assertNotNull(config.getStartSequenceNumbers());
+ Assert.assertNotNull(config.getEndSequenceNumbers());
+ Assert.assertEquals(2, config.getStartSequenceNumbers().size());
+ Assert.assertEquals(2, config.getEndSequenceNumbers().size());
+ }
+
+ @Test
+ public void testDeserializationWithMixedTypes() throws Exception
+ {
+ String json = "{"
+ + "\"startSequenceNumbers\": {\"0\": 100, \"1\": \"200\"},"
+ + "\"endSequenceNumbers\": {\"0\": 500, \"1\": \"600\"}"
+ + "}";
+
+ BoundedStreamConfig config = mapper.readValue(json, BoundedStreamConfig.class);
+
+ Assert.assertNotNull(config.getStartSequenceNumbers());
+ Assert.assertNotNull(config.getEndSequenceNumbers());
+ Assert.assertEquals(2, config.getStartSequenceNumbers().size());
+ Assert.assertEquals(2, config.getEndSequenceNumbers().size());
+ }
+
+ @Test
+ public void testEquals_equalObjects()
+ {
+ Map<String, Long> start1 = new HashMap<>();
+ start1.put("0", 0L);
+ Map<String, Long> end1 = new HashMap<>();
+ end1.put("0", 100L);
+
+ Map<String, Long> start2 = new HashMap<>();
+ start2.put("0", 0L);
+ Map<String, Long> end2 = new HashMap<>();
+ end2.put("0", 100L);
+
+ BoundedStreamConfig config1 = new BoundedStreamConfig(start1, end1);
+ BoundedStreamConfig config2 = new BoundedStreamConfig(start2, end2);
+
+ Assert.assertEquals(config1, config2);
+ Assert.assertEquals(config1.hashCode(), config2.hashCode());
+ }
+
+ @Test
+ public void testEquals_nullObject()
+ {
+ Map<String, Long> start = new HashMap<>();
+ start.put("0", 0L);
+ Map<String, Long> end = new HashMap<>();
+ end.put("0", 100L);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(start, end);
+
+ Assert.assertNotEquals(config, null);
+ }
+
+ @Test
+ public void testEquals_differentClass()
+ {
+ Map<String, Long> start = new HashMap<>();
+ start.put("0", 0L);
+ Map<String, Long> end = new HashMap<>();
+ end.put("0", 100L);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(start, end);
+
+ Assert.assertNotEquals(config, "not a BoundedStreamConfig");
+ }
+
+ @Test
+ public void testEquals_differentStartOffsets()
+ {
+ Map<String, Long> start1 = new HashMap<>();
+ start1.put("0", 0L);
+ Map<String, Long> start2 = new HashMap<>();
+ start2.put("0", 10L);
+ Map<String, Long> end = new HashMap<>();
+ end.put("0", 100L);
+
+ BoundedStreamConfig config1 = new BoundedStreamConfig(start1, end);
+ BoundedStreamConfig config2 = new BoundedStreamConfig(start2, end);
+
+ Assert.assertNotEquals(config1, config2);
+ }
+
+ @Test
+ public void testEquals_differentEndOffsets()
+ {
+ Map<String, Long> start = new HashMap<>();
+ start.put("0", 0L);
+ Map<String, Long> end1 = new HashMap<>();
+ end1.put("0", 100L);
+ Map<String, Long> end2 = new HashMap<>();
+ end2.put("0", 200L);
+
+ BoundedStreamConfig config1 = new BoundedStreamConfig(start, end1);
+ BoundedStreamConfig config2 = new BoundedStreamConfig(start, end2);
+
+ Assert.assertNotEquals(config1, config2);
+ }
+
+ @Test
+ public void testHashCode_consistency()
+ {
+ Map<String, Long> start = new HashMap<>();
+ start.put("0", 0L);
+ Map<String, Long> end = new HashMap<>();
+ end.put("0", 100L);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(start, end);
+
+ int hashCode1 = config.hashCode();
+ int hashCode2 = config.hashCode();
+
+ Assert.assertEquals(hashCode1, hashCode2);
+ }
+
+ @Test
+ public void testHashCode_equalObjectsSameHashCode()
+ {
+ Map<String, Long> start1 = new HashMap<>();
+ start1.put("0", 0L);
+ Map<String, Long> end1 = new HashMap<>();
+ end1.put("0", 100L);
+
+ Map<String, Long> start2 = new HashMap<>();
+ start2.put("0", 0L);
+ Map<String, Long> end2 = new HashMap<>();
+ end2.put("0", 100L);
+
+ BoundedStreamConfig config1 = new BoundedStreamConfig(start1, end1);
+ BoundedStreamConfig config2 = new BoundedStreamConfig(start2, end2);
+
+ Assert.assertEquals(config1.hashCode(), config2.hashCode());
Review Comment:
Could replace it with:
```
EqualsVerifier.forClass(BoundedStreamConfig.class).withNonnullFields("startSequenceNumbers", "endSequenceNumbers").usingGetClass().verify();
```
(Perhaps `EqualsVerifier` tests would replace some of the test permutations for equals/hashcode here)
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * Regression test: a new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED. Before the fix, hasTaskGroupReachedBoundedEnd() compared
+ * the stale committed offset against the new endOffset (e.g. committed=100 >= newEnd=50) and
+ * returned true, bypassing task creation and the documented mismatch error entirely.
+ */
Review Comment:
Regression test or the javadoc can be entirely removed as these are new tests from this PR/self reviews itself:
```suggestion
```
##########
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java:
##########
@@ -381,12 +387,54 @@ protected boolean isShardExpirationMarker(String seqNum)
return KinesisSequenceNumber.EXPIRED_MARKER.equals(seqNum);
}
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ // Kinesis sequence numbers are decimal numeric strings that must be compared numerically.
+ // Use BigInteger because Kinesis sequence numbers can be very large (128-bit).
+ try {
+ BigInteger currentNum = new BigInteger(current);
+ BigInteger targetNum = new BigInteger(target);
+ return currentNum.compareTo(targetNum) >= 0;
+ }
+ catch (NumberFormatException e) {
+ throw new IAE(
+ StringUtils.format(
+ "Invalid Kinesis sequence number. Expected numeric string but got current=[%s], target=[%s]",
+ current,
+ target
+ ),
+ e
+ );
+ }
Review Comment:
Kinesis has some existing utilities for this which also accounts for some special handling of sequence numbers. Could probably be simplified to:
```
return KinesisSequenceNumber.of(current).compareTo(KinesisSequenceNumber.of(target)) >= 0;
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
+ } else {
+ // Inclusive: empty only if start > end
+ isEmpty = isOffsetAtOrBeyond(start, end) && !start.equals(end);
+ }
+
+ if (!isEmpty) {
+ allPartitionsEmptyRange = false;
+ break;
+ }
+ }
+
+ if (allPartitionsEmptyRange) {
+ log.warn(
+ "TaskGroup[%d] has empty range for all partitions (start %s end). "
+ + "No work to do, marking as complete. Start: %s, End: %s",
+ groupId,
+ isEndOffsetExclusive() ? ">=" : ">",
+ startOffsets,
+ endOffsets
+ );
+ return true;
+ }
+
+ // Offsets from a prior bounded run with a different config must not count as completion evidence.
+ // Returning false lets createNewTasks() run, which calls getOffsetFromStorageForPartition()
+ // and throws the documented mismatch error.
+ BoundedStreamConfig metadataBoundedConfig = getBoundedConfigFromMetadata(retrieveDataSourceMetadata());
+ if (!boundedConfig.equals(metadataBoundedConfig)) {
+ return false;
+ }
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ *
+ * @return true if all bounded work is complete, false otherwise
Review Comment:
Thanks for clarifying the docs
nit: for easier navigation between these javadoc functions:
```suggestion
/**
* For bounded supervisors, we determine completion by checking if new tasks would be created.
* In {@link #createNewTasks()}, bounded mode checks {@link #hasTaskGroupReachedBoundedEnd(int)} before creating tasks.
* If that returns true (offsets reached), no new tasks are created.
* So completion is: no active tasks, no pending tasks, and {@link #createNewTasks()} chose not to create any.
*
* @return true if all bounded work is complete, false otherwise
```
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
Review Comment:
Nice, thanks for these tests!
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
Review Comment:
Actually, do you think it's worth resolving this once instead of computing it every time its invoked from the internal loop (given that it doesn't change)?
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
Review Comment:
Perhaps also validate that the records ingested by this bounded supervisor is 0 and/or the tasks spun up by this supervisor actually failed?
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * Regression test: a new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED. Before the fix, hasTaskGroupReachedBoundedEnd() compared
+ * the stale committed offset against the new endOffset (e.g. committed=100 >= newEnd=50) and
+ * returned true, bypassing task creation and the documented mismatch error entirely.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
Review Comment:
Just for completeness, maybe have unequal start / end offset marker for partitions 0 & 1 (in this test or the previous ones)?
##########
docs/ingestion/supervisor.md:
##########
@@ -251,6 +252,57 @@ Before you set `stopTaskCount`, note the following:
- The [task autoscaler](#task-autoscaler) ignores `stopTaskCount` when shutting down tasks in response to a task count change. The task autoscaler needs to redistribute partitions across tasks, which requires all tasks to be shut down.
- If you set `stopTaskCount` to a value less than `taskCount`, Druid cycles the longest running tasks first, then other tasks up to the value set.
+#### Bounded stream configuration
+
+Use `boundedStreamConfig` to configure one-time ingestion from a specific range of offsets. This is useful for backfilling historical data or reprocessing data with different configurations.
+
+The `boundedStreamConfig` object contains the following properties:
+
+|Property|Type|Description|Required|
+|--------|----|-----------|--------|
+|`startSequenceNumbers`|Object|Map of partition IDs to start offsets (inclusive for Kafka, inclusive for Kinesis).|Yes|
+|`endSequenceNumbers`|Object|Map of partition IDs to end offsets (exclusive for Kafka, inclusive for Kinesis).|Yes|
+
+When configured, the supervisor:
+1. Creates tasks that start reading from `startSequenceNumbers`
+2. Tasks automatically stop when they reach `endSequenceNumbers`
+3. Supervisor does not create replacement tasks after tasks complete
+4. Supervisor transitions to `COMPLETED` state and terminates when all tasks finish
+
+**Metadata consistency:** The bounded configuration is stored in datasource metadata along with checkpointed offsets. If you restart the supervisor or create a new supervisor with a different `boundedStreamConfig` for the same datasource, the supervisor will fail with an error. To start a new bounded ingestion with different offsets, either:
+- Use the [supervisor reset API](../api-reference/supervisor-api.md#reset-a-supervisor) to clear existing metadata
+- Use a different datasource name
Review Comment:
Ditto: I think using a different datasource name would already be supported by the multi-supervisor feature (I think we meant different supervisor ID here?)
##########
docs/ingestion/supervisor.md:
##########
@@ -65,6 +65,7 @@ For configuration properties specific to Kafka and Kinesis, see [Kafka I/O confi
|`lateMessageRejectionPeriod`|ISO 8601 period|Configures tasks to reject messages with timestamps earlier than this period before the task was created. For example, if this property is set to `PT1H` and the supervisor creates a task at `2016-01-01T12:00Z`, Druid drops messages with timestamps earlier than `2016-01-01T11:00Z`. This may help prevent concurrency issues if your data stream has late messages and you have multiple pipelines that need to operate on the same segments, such as a streaming and a nightly batch ingestion pipeline. You can specify only one of the late message rejection properties.|No||
|`earlyMessageRejectionPeriod`|ISO 8601 period|Configures tasks to reject messages with timestamps later than this period after the task reached its task duration. For example, if this property is set to `PT1H`, the task duration is set to `PT1H` and the supervisor creates a task at `2016-01-01T12:00Z`, Druid drops messages with timestamps later than `2016-01-01T14:00Z`. Tasks sometimes run past their task duration, such as in cases of supervisor failover.|No||
|`stopTaskCount`|Integer|Limits the number of ingestion tasks Druid can cycle at any given time. If not set, Druid can cycle all tasks at the same time. If set to a value less than `taskCount`, your cluster needs fewer available slots to run the supervisor. You can save costs by scaling down your ingestion tier, but this can lead to slower cycle times and lag. See [`stopTaskCount`](#stoptaskcount) for more information.|No|`taskCount` value|
+|`boundedStreamConfig`|Object|Configures the supervisor for bounded (one-time) ingestion with explicit start and end offsets. When set, the supervisor creates tasks that read from `startSequenceNumbers` to `endSequenceNumbers`, then automatically terminates when all data is ingested. The bounded configuration is stored with datasource metadata; if a supervisor is restarted or a new supervisor is created with different offsets for the same datasource, it will fail. To retry with different offsets, use the supervisor reset API to clear metadata or use a different datasource. Useful for backfills and historical reprocessing. See [Bounded stream configuration](#bounded-stream-configuration) for details.|No|null|
Review Comment:
```suggestion
|`boundedStreamConfig`|Object|Configures the supervisor for bounded (one-time) ingestion with explicit start and end offsets. When set, the supervisor creates tasks that read from `startSequenceNumbers` to `endSequenceNumbers`, then automatically terminates when all data is ingested. The bounded configuration is stored with datasource metadata; if a supervisor is restarted or a new supervisor is created with different offsets for the same datasource, it will fail. To retry with different offsets, use the supervisor reset API to clear metadata or use a different supervisor ID. Useful for backfills and historical reprocessing. See [Bounded stream configuration](#bounded-stream-configuration) for details.|No|null|
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4263,18 +4412,51 @@ private Map<PartitionIdType, OrderedSequenceNumber<SequenceOffsetType>> generate
return builder.build();
}
+ /**
+ * Extracts BoundedStreamConfig from DataSourceMetadata if available.
+ */
+ @Nullable
+ private BoundedStreamConfig getBoundedConfigFromMetadata(@Nullable DataSourceMetadata metadata)
+ {
+ if (metadata instanceof SeekableStreamDataSourceMetadata) {
+ return ((SeekableStreamDataSourceMetadata<?, ?>) metadata).getBoundedStreamConfig();
+ }
+ return null;
+ }
+
/**
* Queries the dataSource metadata table to see if there is a previous ending sequence for this partition. If it
* doesn't find any data, it will retrieve the latest or earliest Kafka/Kinesis sequence depending on the
* {@link SeekableStreamSupervisorIOConfig#useEarliestSequenceNumber}.
*/
private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartition(
PartitionIdType partition,
- final Map<PartitionIdType, SequenceOffsetType> metadataOffsets
+ final Map<PartitionIdType, SequenceOffsetType> metadataOffsets,
+ @Nullable final BoundedStreamConfig metadataBoundedConfig
)
{
SequenceOffsetType sequence = metadataOffsets.get(partition);
if (sequence != null) {
+ // In bounded mode, check if the metadata's bounded config matches the current supervisor's config
+ if (ioConfig.isBounded()) {
+ BoundedStreamConfig currentBoundedConfig = ioConfig.getBoundedStreamConfig();
+
+ // If configs don't match (or metadata has no config), throw exception
+ if (!currentBoundedConfig.equals(metadataBoundedConfig)) {
+ throw DruidException.forPersona(DruidException.Persona.ADMIN)
+ .ofCategory(DruidException.Category.INVALID_INPUT)
+ .build(
+ "Bounded supervisor detected existing metadata from a different run. "
+ + "Metadata bounded config [%s] does not match current config [%s]. "
+ + "To start a new bounded ingestion, either: "
+ + "(1) use the supervisor reset API to clear existing metadata, or "
+ + "(2) use a different supervisor ID / datasource.",
Review Comment:
I think just a different supervisor ID would suffice for this backfill requirement? A different datasource would already be supported
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
Review Comment:
Have an early return before the convert* methods are called?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
+ } else {
+ // Inclusive: empty only if start > end
+ isEmpty = isOffsetAtOrBeyond(start, end) && !start.equals(end);
+ }
+
+ if (!isEmpty) {
+ allPartitionsEmptyRange = false;
+ break;
+ }
+ }
+
+ if (allPartitionsEmptyRange) {
+ log.warn(
+ "TaskGroup[%d] has empty range for all partitions (start %s end). "
+ + "No work to do, marking as complete. Start: %s, End: %s",
+ groupId,
+ isEndOffsetExclusive() ? ">=" : ">",
+ startOffsets,
+ endOffsets
+ );
+ return true;
+ }
+
+ // Offsets from a prior bounded run with a different config must not count as completion evidence.
+ // Returning false lets createNewTasks() run, which calls getOffsetFromStorageForPartition()
+ // and throws the documented mismatch error.
Review Comment:
Not sure what this means without enough context: `// and throws the documented mismatch error.`
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * Regression test: a new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED. Before the fix, hasTaskGroupReachedBoundedEnd() compared
+ * the stale committed offset against the new endOffset (e.g. committed=100 >= newEnd=50) and
+ * returned true, bypassing task creation and the documented mismatch error entirely.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
+
+ cluster.callApi().postSupervisor(supervisor1);
+ waitUntilPublishedRecordsAreIngested(200);
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Run 2: same datasource, endOffset (50) < stale committed offset (100).
+ // Without the fix the supervisor reaches COMPLETED immediately without running tasks.
+ // With the fix it detects the config mismatch and becomes UNHEALTHY_SUPERVISOR.
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 0L);
+ startOffsets2.put("1", 0L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 50L);
+ endOffsets2.put("1", 50L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig2);
+
+ cluster.callApi().postSupervisor(supervisor2);
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
Review Comment:
Do you think validating a third supervisor with a different ID but the same bounds would be worth adding? (either this test or the following one)
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
Review Comment:
nit: could we make variables final here & in surrounding code where applicable so the intent is clear and nothing accidentally mutates these offsets (except `allPartitionsEmptyRange`)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3237467550
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,182 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
Review Comment:
Yeah good point, I think that would be more optimal. But since these are relatively cheap anyway I'll defer this change for now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3164941305
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
Review Comment:
Ahh, let me update the javadoc to be more clear. The implementation does take that scenario into account. Checking active/pending task group is just a cheap check to avoid hitting the metadata store every run. But if both are empty it will check the offsets in the metadata store as the final validation that it is complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3237433763
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * A new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 150L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
+
+ cluster.callApi().postSupervisor(supervisor1);
+ waitUntilPublishedRecordsAreIngested(250);
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Run 2: same datasource, endOffset (50) < stale committed offset (100).
+ // Without the fix the supervisor reaches COMPLETED immediately without running tasks.
+ // With the fix it detects the config mismatch and becomes UNHEALTHY_SUPERVISOR.
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 0L);
+ startOffsets2.put("1", 0L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 50L);
+ endOffsets2.put("1", 50L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig2);
+
+ cluster.callApi().postSupervisor(supervisor2);
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ private void waitForSupervisorToComplete(String supervisorId)
+ {
+ // Wait for supervisor to reach COMPLETED state
+ int maxAttempts = 60; // 60 seconds timeout
+ int attempt = 0;
+
+ while (attempt < maxAttempts) {
+ try {
+ SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisorId);
+ if ("COMPLETED".equals(status.getState())) {
+ return;
+ }
+ Thread.sleep(1000);
+ attempt++;
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for supervisor to complete", e);
+ }
+ catch (Exception e) {
+ // Supervisor might not be found immediately, retry
+ attempt++;
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting", ie);
+ }
+ }
+ }
+
+ Assertions.fail("Supervisor did not complete within timeout");
+ }
+
+ private void waitForSupervisorToBeUnhealthy(String supervisorId)
+ {
+ // Wait for supervisor to become unhealthy after detecting the metadata mismatch
+ int maxAttempts = 30; // 30 seconds timeout
+ int attempt = 0;
+
+ while (attempt < maxAttempts) {
+ try {
+ SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisorId);
+
+ // The supervisor should become unhealthy when the exception is thrown
+ if (!status.isHealthy()) {
+ log.info("Supervisor became unhealthy with state: %s", status.getDetailedState());
+ return;
+ }
+
+ Thread.sleep(1000);
+ attempt++;
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for supervisor to become unhealthy", e);
+ }
+ catch (Exception e) {
+ // Supervisor might not be found immediately, retry
+ attempt++;
Review Comment:
Thanks for the suggestion. That is much cleaner, I went ahead and made that update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3208935781
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,175 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
+ } else {
+ // Inclusive: empty only if start > end
+ isEmpty = isOffsetAtOrBeyond(start, end) && !start.equals(end);
+ }
+
+ if (!isEmpty) {
+ allPartitionsEmptyRange = false;
+ break;
+ }
+ }
+
+ if (allPartitionsEmptyRange) {
+ log.warn(
+ "TaskGroup[%d] has empty range for all partitions (start %s end). "
+ + "No work to do, marking as complete. Start: %s, End: %s",
+ groupId,
+ isEndOffsetExclusive() ? ">=" : ">",
+ startOffsets,
+ endOffsets
+ );
+ return true;
+ }
+
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
Review Comment:
Thanks, the current update addresses my concern: `hasTaskGroupReachedBoundedEnd()` now refuses to treat stored offsets as completion evidence when the metadata bounded config differs, so task creation can reach the existing mismatch error path. The added stale-offset regression test covers the case I was worried about.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3230872042
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
Review Comment:
In the case of metadata mismatch the tasks are never actually spun up. An exception is thrown before tasks are ever created.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3231378733
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * A new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 150L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
+
+ cluster.callApi().postSupervisor(supervisor1);
+ waitUntilPublishedRecordsAreIngested(250);
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Run 2: same datasource, endOffset (50) < stale committed offset (100).
+ // Without the fix the supervisor reaches COMPLETED immediately without running tasks.
+ // With the fix it detects the config mismatch and becomes UNHEALTHY_SUPERVISOR.
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 0L);
+ startOffsets2.put("1", 0L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 50L);
+ endOffsets2.put("1", 50L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig2);
+
+ cluster.callApi().postSupervisor(supervisor2);
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ private void waitForSupervisorToComplete(String supervisorId)
+ {
+ // Wait for supervisor to reach COMPLETED state
+ int maxAttempts = 60; // 60 seconds timeout
+ int attempt = 0;
+
+ while (attempt < maxAttempts) {
+ try {
+ SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisorId);
+ if ("COMPLETED".equals(status.getState())) {
+ return;
+ }
+ Thread.sleep(1000);
+ attempt++;
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for supervisor to complete", e);
+ }
+ catch (Exception e) {
+ // Supervisor might not be found immediately, retry
+ attempt++;
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting", ie);
+ }
+ }
+ }
+
+ Assertions.fail("Supervisor did not complete within timeout");
+ }
+
+ private void waitForSupervisorToBeUnhealthy(String supervisorId)
+ {
+ // Wait for supervisor to become unhealthy after detecting the metadata mismatch
+ int maxAttempts = 30; // 30 seconds timeout
+ int attempt = 0;
+
+ while (attempt < maxAttempts) {
+ try {
+ SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisorId);
+
+ // The supervisor should become unhealthy when the exception is thrown
+ if (!status.isHealthy()) {
+ log.info("Supervisor became unhealthy with state: %s", status.getDetailedState());
+ return;
+ }
+
+ Thread.sleep(1000);
+ attempt++;
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for supervisor to become unhealthy", e);
+ }
+ catch (Exception e) {
+ // Supervisor might not be found immediately, retry
+ attempt++;
Review Comment:
Same comment - what do you think about enabling the `SupervisorStatsMonitor` to the base class's monitors? I think it’d be cleaner and more resilient to latch on the `supervisor/count` metric emitted by it:
```java
.addCommonProperty("druid.monitoring.emissionPeriod", "PT1s")
.addCommonProperty(
"druid.monitoring.monitors",
"[\"org.apache.druid.server.metrics.SupervisorStatsMonitor\"]"
)
```
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, currentOffsets.size(), "Should have 2 partitions");
+
+ // Create first bounded config - ingest only the first 100 records from each partition
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 100L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+
+ // Create first bounded supervisor and run it to completion
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig1
+ );
+
+ cluster.callApi().postSupervisor(supervisor1);
+
+ // Wait for records to be ingested (approximately 200 records total from both partitions)
+ waitUntilPublishedRecordsAreIngested(200);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Now try to create a second bounded supervisor with different bounded config on the same datasource
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 50L); // Different start offset
+ startOffsets2.put("1", 50L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 200L); // Different end offset
+ endOffsets2.put("1", 200L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig2
+ );
+
+ // Post the second supervisor (it should use the same supervisor ID/datasource)
+ cluster.callApi().postSupervisor(supervisor2);
+
+ // Wait for the supervisor to process and detect the metadata mismatch
+ // The exception we're testing for is thrown and logged, and causes the supervisor to become unhealthy
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ // Verify the supervisor is unhealthy
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ /**
+ * A new bounded run whose endOffset is less than the offset committed by a prior
+ * run must not silently reach COMPLETED.
+ */
+ @Test
+ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceedsNewEnd()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+ publish1kRecords(topic, false);
+
+ // Run 1: ingest up to offset 100 on each partition and complete.
+ Map<String, Long> startOffsets1 = new HashMap<>();
+ startOffsets1.put("0", 0L);
+ startOffsets1.put("1", 0L);
+
+ Map<String, Long> endOffsets1 = new HashMap<>();
+ endOffsets1.put("0", 100L);
+ endOffsets1.put("1", 150L);
+
+ BoundedStreamConfig boundedConfig1 = new BoundedStreamConfig(startOffsets1, endOffsets1);
+ final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
+
+ cluster.callApi().postSupervisor(supervisor1);
+ waitUntilPublishedRecordsAreIngested(250);
+ waitForSupervisorToComplete(supervisor1.getId());
+
+ final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());
+ Assertions.assertEquals("COMPLETED", status1.getState());
+
+ // Run 2: same datasource, endOffset (50) < stale committed offset (100).
+ // Without the fix the supervisor reaches COMPLETED immediately without running tasks.
+ // With the fix it detects the config mismatch and becomes UNHEALTHY_SUPERVISOR.
+ Map<String, Long> startOffsets2 = new HashMap<>();
+ startOffsets2.put("0", 0L);
+ startOffsets2.put("1", 0L);
+
+ Map<String, Long> endOffsets2 = new HashMap<>();
+ endOffsets2.put("0", 50L);
+ endOffsets2.put("1", 50L);
+
+ BoundedStreamConfig boundedConfig2 = new BoundedStreamConfig(startOffsets2, endOffsets2);
+ final KafkaSupervisorSpec supervisor2 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig2);
+
+ cluster.callApi().postSupervisor(supervisor2);
+ waitForSupervisorToBeUnhealthy(supervisor2.getId());
+
+ final SupervisorStatus status2 = cluster.callApi().getSupervisorStatus(supervisor2.getId());
+ Assertions.assertFalse(status2.isHealthy(), "Supervisor should be unhealthy after detecting metadata mismatch");
+ Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(), "Supervisor state should be UNHEALTHY_SUPERVISOR");
+ }
+
+ private void waitForSupervisorToComplete(String supervisorId)
+ {
+ // Wait for supervisor to reach COMPLETED state
+ int maxAttempts = 60; // 60 seconds timeout
+ int attempt = 0;
+
+ while (attempt < maxAttempts) {
+ try {
+ SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisorId);
+ if ("COMPLETED".equals(status.getState())) {
+ return;
+ }
+ Thread.sleep(1000);
+ attempt++;
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for supervisor to complete", e);
+ }
+ catch (Exception e) {
+ // Supervisor might not be found immediately, retry
+ attempt++;
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting", ie);
Review Comment:
I think we should be able to latch on this metric instead of retry-with-sleeps:
```java
overlord.latchableEmitter().waitForEvent(
event -> event.hasMetricName("supervisor/count")
.hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
.hasDimension("state", "COMPLETED")
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3192471603
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java:
##########
@@ -1529,4 +1534,74 @@
return autoScalerConfig;
}
+ @Test
+ public void testBoundedStreamSupervisorSpec_runsWithBoundedConfig()
+ {
+ EmittingLogger.registerEmitter(emitter);
+
+ Map<String, Object> startOffsets = ImmutableMap.of("0", 0L);
+ Map<String, Object> endOffsets = ImmutableMap.of("0", 100L);
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ SeekableStreamSupervisorIOConfig boundedIoConfig = new SeekableStreamSupervisorIOConfig(
+ "stream",
+ new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
+ 1,
+ 1,
+ new Period("PT1H"),
+ new Period("P1D"),
+ new Period("PT30S"),
+ false,
+ new Period("PT30M"),
+ null,
+ null,
+ null,
+ LagAggregator.DEFAULT,
+ null,
+ null,
+ null,
+ null,
+ boundedConfig
+ )
+ {
+ };
+
+ EasyMock.expect(spec.getId()).andReturn(SUPERVISOR).anyTimes();
+ EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
+ EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
Review Comment:
## CodeQL / Deprecated method or constructor invocation
Invoking [SeekableStreamSupervisorSpec.getDataSchema](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/11176)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 merged PR #19372:
URL: https://github.com/apache/druid/pull/19372
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3165031704
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
+ */
+ private boolean isBoundedWorkComplete()
+ {
+ if (!ioConfig.isBounded()) {
+ return false;
+ }
+
+ // Check if task groups are empty (no tasks active or pending)
+ boolean noActiveTasks = activelyReadingTaskGroups.isEmpty();
+ boolean noPendingTasks = pendingCompletionTaskGroups.values().stream().allMatch(List::isEmpty);
+
+ if (!noActiveTasks || !noPendingTasks) {
+ return false;
+ }
+
+ // At this point, no tasks are running. Since createNewTasks() already ran,
+ // if tasks aren't running it means either:
+ // A) Tasks completed successfully and offset targets were reached (don't recreate)
+ // B) Tasks failed/killed and haven't reached targets (will recreate next run)
+ //
+ // To distinguish, we check if createNewTasks() would create new tasks.
+ // If hasTaskGroupReachedBoundedEnd() returns false for any group, createNewTasks()
+ // will create tasks next iteration, so we're not complete.
+ for (Integer groupId : partitionGroups.keySet()) {
+ if (!hasTaskGroupReachedBoundedEnd(groupId)) {
+ log.debug("TaskGroup[%d] has not reached bounded end, tasks will be recreated", groupId);
+ return false;
+ }
+ }
+
+ // All groups have reached their end offsets and no tasks are running.
+ // Work is complete!
+ log.info("All bounded tasks completed for supervisor[%s]", supervisorId);
+ return true;
+ }
+
+ /**
+ * Handle bounded processing completion by shutting down the supervisor.
+ * At this point, all task groups are already empty (verified by isBoundedWorkComplete),
+ * so we just need to mark the supervisor as completed.
+ */
+ private void handleBoundedCompletion()
+ {
+ log.info("Bounded processing complete for supervisor[%s]. Marking as COMPLETED.", supervisorId);
+ stateManager.maybeSetState(SupervisorStateManager.BasicState.COMPLETED);
Review Comment:
The latest [commit](https://github.com/apache/druid/pull/19372/changes/4bed6580f659959873a01b1ea926bb8c8c96fa26#diff-ee5ec6696f9263bed19eefcb5f24e25b4141d15aaa4f13853e8790227e52258b) handles this workflow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3168201885
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
I agree the source of the metadata is the hard part here, but I would avoid using "metadata is within the configured start/end range" as the deciding rule. If the stored offset is inside `[start, end)`, starting there can still silently skip the prefix `[start, storedOffset)`, which is the same class of surprise as the original issue. I think the safer behavior is still to fail fast when bounded mode finds existing metadata for the configured bounded partitions unless there is an explicit signal that this is a resume of the same bounded run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3203825756
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,175 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
+ } else {
+ // Inclusive: empty only if start > end
+ isEmpty = isOffsetAtOrBeyond(start, end) && !start.equals(end);
+ }
+
+ if (!isEmpty) {
+ allPartitionsEmptyRange = false;
+ break;
+ }
+ }
+
+ if (allPartitionsEmptyRange) {
+ log.warn(
+ "TaskGroup[%d] has empty range for all partitions (start %s end). "
+ + "No work to do, marking as complete. Start: %s, End: %s",
+ groupId,
+ isEndOffsetExclusive() ? ">=" : ">",
+ startOffsets,
+ endOffsets
+ );
+ return true;
+ }
+
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
Review Comment:
Thanks @FrankChen021 I did leave that as a comment in the [embedded test](https://github.com/apache/druid/pull/19372/changes#diff-0a11abc4cce2bf91968fdffd9cb4cbde67428de4b4f7541101a7b2ef7f34b35cR185-R189)
I suppose it would be better to be explicit about this mismatch here. Will make that update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3181979286
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
Thanks, this looks like the right direction to me. Persisting the bounded config in the datasource metadata and rejecting existing metadata whose bounded config is missing or different addresses the silent stale-offset case, while still allowing a supervisor to resume metadata from the same bounded run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3156717978
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
+ */
+ private boolean isBoundedWorkComplete()
+ {
+ if (!ioConfig.isBounded()) {
+ return false;
+ }
+
+ // Check if task groups are empty (no tasks active or pending)
+ boolean noActiveTasks = activelyReadingTaskGroups.isEmpty();
+ boolean noPendingTasks = pendingCompletionTaskGroups.values().stream().allMatch(List::isEmpty);
+
+ if (!noActiveTasks || !noPendingTasks) {
+ return false;
+ }
+
+ // At this point, no tasks are running. Since createNewTasks() already ran,
+ // if tasks aren't running it means either:
+ // A) Tasks completed successfully and offset targets were reached (don't recreate)
+ // B) Tasks failed/killed and haven't reached targets (will recreate next run)
+ //
+ // To distinguish, we check if createNewTasks() would create new tasks.
+ // If hasTaskGroupReachedBoundedEnd() returns false for any group, createNewTasks()
+ // will create tasks next iteration, so we're not complete.
+ for (Integer groupId : partitionGroups.keySet()) {
+ if (!hasTaskGroupReachedBoundedEnd(groupId)) {
+ log.debug("TaskGroup[%d] has not reached bounded end, tasks will be recreated", groupId);
+ return false;
+ }
+ }
+
+ // All groups have reached their end offsets and no tasks are running.
+ // Work is complete!
+ log.info("All bounded tasks completed for supervisor[%s]", supervisorId);
+ return true;
+ }
+
+ /**
+ * Handle bounded processing completion by shutting down the supervisor.
+ * At this point, all task groups are already empty (verified by isBoundedWorkComplete),
+ * so we just need to mark the supervisor as completed.
+ */
+ private void handleBoundedCompletion()
+ {
+ log.info("Bounded processing complete for supervisor[%s]. Marking as COMPLETED.", supervisorId);
+ stateManager.maybeSetState(SupervisorStateManager.BasicState.COMPLETED);
Review Comment:
Should this call `stop()` with this COMPLETED state so things get unregistered and the executor is removed?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -259,6 +262,42 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
if (!this.getSource().equals(other.getSource())) {
throw InvalidInput.exception(ILLEGAL_INPUT_SOURCE_UPDATE_ERROR_MESSAGE, this.getSource(), other.getSource());
}
+
+ // Validate bounded stream configuration
+ validateBoundedStreamConfig(other);
Review Comment:
Looks like this is only invoked on an update. Should we validate this during creation time as well?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -259,6 +262,42 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
if (!this.getSource().equals(other.getSource())) {
throw InvalidInput.exception(ILLEGAL_INPUT_SOURCE_UPDATE_ERROR_MESSAGE, this.getSource(), other.getSource());
}
+
+ // Validate bounded stream configuration
+ validateBoundedStreamConfig(other);
+ }
+
+ /**
+ * Validates bounded stream configuration for the supervisor spec.
+ *
+ * @param spec the supervisor spec to validate
+ * @throws DruidException if the bounded stream configuration is invalid
+ */
+ protected void validateBoundedStreamConfig(SeekableStreamSupervisorSpec spec) throws DruidException
+ {
+ SeekableStreamSupervisorIOConfig ioConfig = spec.getIoConfig();
+
+ if (ioConfig.isBounded()) {
+ // Validate partition consistency
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ if (!boundedConfig.getStartSequenceNumbers().keySet().equals(boundedConfig.getEndSequenceNumbers().keySet())) {
+ throw InvalidInput.exception(
+ "Bounded stream config has mismatched partitions. Start: %s, End: %s",
+ boundedConfig.getStartSequenceNumbers().keySet(),
+ boundedConfig.getEndSequenceNumbers().keySet()
+ );
+ }
Review Comment:
I wonder if some of these checks during create & update flows can be shared. Perhaps add a static helper inside `BoundedStreamConfig` that can be validated by both create & `validateSpecUpdateTo`?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
Review Comment:
Hmm, indirectly inferring bounded work completion from the absence of tasks seems error-prone since these things can race and function independently in some ways.
For example, if all tasks failed or if tasks were queued but haven’t started yet as mentioned in point 3, would the supervisor transition to a completed state prematurely?
I think a more foolproof way to check completion would be to directly check metadata storage to track offsets for work completion and have the design reflect that with appropriate guardrails. I believe this also relates to the discussion here: https://github.com/apache/druid/pull/19372/changes#r3141594283
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfigTest.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class BoundedStreamConfigTest
+{
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ public void testConstructorWithValidMaps()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+ startOffsets.put("1", 200L);
+
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500L);
+ endOffsets.put("1", 600L);
+
+ BoundedStreamConfig config = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ Assert.assertEquals(startOffsets, config.getStartSequenceNumbers());
+ Assert.assertEquals(endOffsets, config.getEndSequenceNumbers());
+ }
+
+ @Test
+ public void testConstructorWithNullStartSequenceNumbers()
+ {
+ Map<String, Long> endOffsets = new HashMap<>();
+ endOffsets.put("0", 500L);
+
+ NullPointerException ex = Assert.assertThrows(
+ NullPointerException.class,
+ () -> new BoundedStreamConfig(null, endOffsets)
+ );
+
+ Assert.assertTrue(ex.getMessage().contains("startSequenceNumbers"));
+ }
+
+ @Test
+ public void testConstructorWithNullEndSequenceNumbers()
+ {
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 100L);
+
+ NullPointerException ex = Assert.assertThrows(
+ NullPointerException.class,
+ () -> new BoundedStreamConfig(startOffsets, null)
+ );
Review Comment:
Seems strange to assert on NPEs. Should we have defensive code that throws DruidExceptions instead of NPEs wtih preconditions?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -259,6 +262,42 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
if (!this.getSource().equals(other.getSource())) {
throw InvalidInput.exception(ILLEGAL_INPUT_SOURCE_UPDATE_ERROR_MESSAGE, this.getSource(), other.getSource());
}
+
+ // Validate bounded stream configuration
+ validateBoundedStreamConfig(other);
+ }
+
+ /**
+ * Validates bounded stream configuration for the supervisor spec.
+ *
+ * @param spec the supervisor spec to validate
+ * @throws DruidException if the bounded stream configuration is invalid
+ */
+ protected void validateBoundedStreamConfig(SeekableStreamSupervisorSpec spec) throws DruidException
+ {
+ SeekableStreamSupervisorIOConfig ioConfig = spec.getIoConfig();
+
+ if (ioConfig.isBounded()) {
+ // Validate partition consistency
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ if (!boundedConfig.getStartSequenceNumbers().keySet().equals(boundedConfig.getEndSequenceNumbers().keySet())) {
+ throw InvalidInput.exception(
+ "Bounded stream config has mismatched partitions. Start: %s, End: %s",
+ boundedConfig.getStartSequenceNumbers().keySet(),
+ boundedConfig.getEndSequenceNumbers().keySet()
+ );
+ }
+
+ // Warn if useConcurrentLocks is not enabled
+ Map<String, Object> context = spec.getContext();
+ if (context == null || !Boolean.TRUE.equals(context.get("useConcurrentLocks"))) {
+ log.warn(
+ "Bounded stream processing without 'useConcurrentLocks=true' may fail " +
+ "if other supervisors are running or segments already exist for these intervals. " +
+ "Consider setting useConcurrentLocks=true in the supervisor context."
+ );
Review Comment:
Is this requirement needed in the general case when an operator is submitting a supervisor with these bounds? I’d expect `useConcurrentLocks` to be required only when a backfill supervisor is submitted in parallel with resetting the currently running supervisor. If that’s the case, would it make sense to enforce this at the reset and backfill API layer instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3141594283
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
P1 Bounded starts can be ignored when metadata exists
getOffsetFromStorageForPartition only falls back to boundedStreamConfig.startSequenceNumbers when no metadata/checkpoint offset exists. If a supervisor is reset or reconfigured with a requested bounded start while metadata storage still has an older offset, the stored metadata wins and the task starts from the stale position instead of the user-supplied bounded start. That can skip the requested backfill range or process a different interval than configured. Bounded mode should either clear/namespace old metadata for the run or explicitly prefer the configured start when initializing the bounded task group.
##########
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java:
##########
@@ -381,6 +381,28 @@ protected boolean isShardExpirationMarker(String seqNum)
return KinesisSequenceNumber.EXPIRED_MARKER.equals(seqNum);
}
+ @Override
+ protected boolean isOffsetAtOrBeyond(String current, String target)
+ {
+ // Kinesis sequence numbers are comparable strings
+ // They can be compared lexicographically to determine order
+ return current.compareTo(target) >= 0;
+ }
+
+ @Override
+ protected String createPartitionIdFromString(String partitionIdString)
+ {
+ // Kinesis uses String as partition ID, so just return the string as-is
+ return partitionIdString;
+ }
+
Review Comment:
P1 Kinesis bounded completion compares offsets lexicographically
isOffsetAtOrBeyond uses current.compareTo(target), which compares Kinesis sequence numbers lexicographically. Kinesis sequence numbers are decimal strings whose numeric order is not generally the same as lexicographic order, especially across different digit lengths. A bounded supervisor can mark a shard complete before the numeric end offset is reached, or keep recreating tasks after it passed the end. Compare Kinesis offsets using their numeric value or the existing Kinesis sequence-number ordering semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3149708894
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorStateManager.java:
##########
@@ -64,7 +64,8 @@ public enum BasicState implements State
RUNNING(true, false),
IDLE(true, false),
SUSPENDED(true, false),
- STOPPING(true, false);
+ STOPPING(true, false),
+ COMPLETED(true, false);
Review Comment:
Yeah if tasks are failing the Supervisor would enter an `UNHEALTHY` state. It would only transition to `COMPLETED` after the specified offset range was successfully processed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3154602223
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
That validation approach sounds right to me. The important part is preventing a bounded supervisor from accidentally mixing explicit bounded start offsets with existing committed metadata for the same id; failing fast with guidance to use a different id or reset the supervisor would avoid the silent stale-offset behavior without deleting or overwriting a running supervisor's offsets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3149665533
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/BoundedStreamConfig.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.seekablestream.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Configuration for bounded (one-time) stream processing with explicit start/end offsets.
Review Comment:
Start offsets are inclusive, but end offsets depend on the implementation for isEndOffsetExclusive. For kinesis it is inclusive, but for Kafka and Rabbit it is exclusive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3175998966
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
Thanks @FrankChen021! I took a stab at this in the most recent [commit](https://github.com/apache/druid/pull/19372/changes/5965ac4de5f1e344906ddbe1a94afce29b85b04e)
Please let me know your thoughts when you get the chance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on PR #19372:
URL: https://github.com/apache/druid/pull/19372#issuecomment-4445923808
Thanks for all the reviews on this one @abhishekrb19 @FrankChen021 I'll go ahead and merge this one to checkpoint the work. Let me know if you spot anything else, and I'll include it in a follow up PR along with the resetOffsetAndBackfill change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3234376992
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4377,6 +4577,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
Review Comment:
[P2] Reject reversed bounded ranges instead of completing
For exclusive streams, this treats `start > end` the same as the valid empty case `start == end`, so a config like `start=500/end=100` makes every partition look complete and the supervisor can move to COMPLETED without creating a task. `KafkaIndexTaskIOConfig` would reject `end < start` if a task were created, but this completion check bypasses that validation. Reversed ranges should fail validation instead of being reported as successful bounded ingestion.
##########
extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java:
##########
@@ -4725,6 +4730,201 @@ public void test_doesTaskMatchSupervisor()
Assert.assertFalse(supervisor.doesTaskMatchSupervisor(differentTaskType));
}
+ @Test
+ public void testBoundedModeCreateTasksWithCorrectOffsets()
+ {
+ Map<String, Object> startOffsets = ImmutableMap.of(
+ "shardId-000000000000", "49590338271490256608559692538361571095921575989136588898",
+ "shardId-000000000001", "49590338271512257353759162668991891722121171891717232706"
+ );
+ Map<String, Object> endOffsets = ImmutableMap.of(
+ "shardId-000000000000", "49590338271534258098958632799622211348320767794297876514",
+ "shardId-000000000001", "49590338271556258844158102930252531974520363696878520322"
+ );
+ final KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
+ STREAM,
+ INPUT_FORMAT,
+ "awsEndpoint",
+ null,
+ 1,
+ 1,
+ new Period("PT30S"),
+ null,
+ new Period("PT30M"),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 0,
+ null,
+ null,
+ null,
+ true,
+ null,
+ new BoundedStreamConfig(startOffsets, endOffsets)
+ );
+
+ Assert.assertTrue(kinesisSupervisorIOConfig.isBounded());
+
+ final KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(null, null);
+ final KinesisSupervisorSpec spec = new KinesisSupervisorSpec(
+ null,
+ null,
+ dataSchema,
+ KinesisSupervisorTuningConfig.defaultConfig(),
+ kinesisSupervisorIOConfig,
+ null,
+ false,
+ taskStorage,
+ taskMaster,
+ indexerMetadataStorageCoordinator,
+ taskClientFactory,
+ OBJECT_MAPPER,
+ new NoopServiceEmitter(),
+ new DruidMonitorSchedulerConfig(),
+ rowIngestionMetersFactory,
+ null,
+ new SupervisorStateManagerConfig()
+ );
+
+ supervisor = new TestableKinesisSupervisor(
+ taskStorage,
+ taskMaster,
+ indexerMetadataStorageCoordinator,
+ taskClientFactory,
+ OBJECT_MAPPER,
+ spec,
+ rowIngestionMetersFactory
+ );
+
+ // Test type conversion methods
+ String shardId = supervisor.createPartitionIdFromString("shardId-000000000000");
+ Assert.assertEquals("shardId-000000000000", shardId);
+
+ String offset = supervisor.createSequenceOffsetFromObject("49590338271490256608559692538361571095921575989136588898");
+ Assert.assertEquals("49590338271490256608559692538361571095921575989136588898", offset);
+
+ offset = supervisor.createSequenceOffsetFromObject(100);
+ Assert.assertEquals("100", offset);
+
+ Assert.assertTrue(supervisor.isOffsetAtOrBeyond(
+ "49590338271512257353759162668991891722121171891717232706",
+ "49590338271490256608559692538361571095921575989136588898"
+ ));
+ Assert.assertTrue(supervisor.isOffsetAtOrBeyond(
+ "49590338271490256608559692538361571095921575989136588898",
+ "49590338271490256608559692538361571095921575989136588898"
+ ));
+ Assert.assertFalse(supervisor.isOffsetAtOrBeyond(
+ "49590338271490256608559692538361571095921575989136588898",
+ "49590338271512257353759162668991891722121171891717232706"
+ ));
+ }
+
+ @Test
+ public void testIsOffsetAtOrBeyond_invalidSequenceNumber()
+ {
+ supervisor = getTestableSupervisor(1, 1, true, "PT1H", null, null);
+
+ Exception e = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> supervisor.isOffsetAtOrBeyond("not-a-number", "12345")
+ );
+ Assert.assertTrue(e.getMessage().contains("Invalid Kinesis sequence number"));
Review Comment:
[P1] Fix invalid sequence assertion
This new assertion expects the exception message to contain `Invalid Kinesis sequence number`, but `KinesisSupervisor.isOffsetAtOrBeyond` delegates to `KinesisSequenceNumber.of`, which constructs a `BigInteger` and throws `NumberFormatException` with a different message for `not-a-number`. As written, this test should fail; either wrap invalid sequence numbers with the expected message in the production method or relax the assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3164946199
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -259,6 +262,42 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
if (!this.getSource().equals(other.getSource())) {
throw InvalidInput.exception(ILLEGAL_INPUT_SOURCE_UPDATE_ERROR_MESSAGE, this.getSource(), other.getSource());
}
+
+ // Validate bounded stream configuration
+ validateBoundedStreamConfig(other);
+ }
+
+ /**
+ * Validates bounded stream configuration for the supervisor spec.
+ *
+ * @param spec the supervisor spec to validate
+ * @throws DruidException if the bounded stream configuration is invalid
+ */
+ protected void validateBoundedStreamConfig(SeekableStreamSupervisorSpec spec) throws DruidException
+ {
+ SeekableStreamSupervisorIOConfig ioConfig = spec.getIoConfig();
+
+ if (ioConfig.isBounded()) {
+ // Validate partition consistency
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ if (!boundedConfig.getStartSequenceNumbers().keySet().equals(boundedConfig.getEndSequenceNumbers().keySet())) {
+ throw InvalidInput.exception(
+ "Bounded stream config has mismatched partitions. Start: %s, End: %s",
+ boundedConfig.getStartSequenceNumbers().keySet(),
+ boundedConfig.getEndSequenceNumbers().keySet()
+ );
+ }
+
+ // Warn if useConcurrentLocks is not enabled
+ Map<String, Object> context = spec.getContext();
+ if (context == null || !Boolean.TRUE.equals(context.get("useConcurrentLocks"))) {
+ log.warn(
+ "Bounded stream processing without 'useConcurrentLocks=true' may fail " +
+ "if other supervisors are running or segments already exist for these intervals. " +
+ "Consider setting useConcurrentLocks=true in the supervisor context."
+ );
Review Comment:
Yeah, this is just logged as a warning and is not a strict requirement. In the reset and backfill API that would be a strict enforcement
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3150398648
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4255,6 +4418,23 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
}
Review Comment:
Thanks for the review @FrankChen021!
> Bounded mode should either clear/namespace old metadata for the run
I'm a bit wary of automatic cleanup of metadata. I'm thinking through the scenario where a cluster operator has a running Supervisor. They want to re-ingest some older data so they resubmit the exact same spec (forgetting to update `id`) so the Bounded Supervisor succeeds but the previously committed offset gets lost.
> explicitly prefer the configured start when initializing the bounded task group.
This falls into the same issue as above where if the operator forgets to set the `id` to something different than the running Supervisor then the previous committed offset is lost forever
I'm leaning towards adding validation that if metadata already exists for the `id` then just throw an exception and suggest the operator to resubmit with a different `id` or reset the Supervisor. Curious to hear your thoughts on this approach. Thanks again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3164978799
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
+ */
+ private boolean isBoundedWorkComplete()
+ {
+ if (!ioConfig.isBounded()) {
+ return false;
+ }
+
+ // Check if task groups are empty (no tasks active or pending)
+ boolean noActiveTasks = activelyReadingTaskGroups.isEmpty();
+ boolean noPendingTasks = pendingCompletionTaskGroups.values().stream().allMatch(List::isEmpty);
+
+ if (!noActiveTasks || !noPendingTasks) {
+ return false;
+ }
+
+ // At this point, no tasks are running. Since createNewTasks() already ran,
+ // if tasks aren't running it means either:
+ // A) Tasks completed successfully and offset targets were reached (don't recreate)
+ // B) Tasks failed/killed and haven't reached targets (will recreate next run)
+ //
+ // To distinguish, we check if createNewTasks() would create new tasks.
+ // If hasTaskGroupReachedBoundedEnd() returns false for any group, createNewTasks()
+ // will create tasks next iteration, so we're not complete.
+ for (Integer groupId : partitionGroups.keySet()) {
+ if (!hasTaskGroupReachedBoundedEnd(groupId)) {
+ log.debug("TaskGroup[%d] has not reached bounded end, tasks will be recreated", groupId);
+ return false;
+ }
+ }
+
+ // All groups have reached their end offsets and no tasks are running.
+ // Work is complete!
+ log.info("All bounded tasks completed for supervisor[%s]", supervisorId);
+ return true;
+ }
+
+ /**
+ * Handle bounded processing completion by shutting down the supervisor.
+ * At this point, all task groups are already empty (verified by isBoundedWorkComplete),
+ * so we just need to mark the supervisor as completed.
+ */
+ private void handleBoundedCompletion()
+ {
+ log.info("Bounded processing complete for supervisor[%s]. Marking as COMPLETED.", supervisorId);
+ stateManager.maybeSetState(SupervisorStateManager.BasicState.COMPLETED);
Review Comment:
One workflow I was testing out was to submit a bounded Supervisor and have it run to completion. Then I adjusted the start/end offsets and re-submitted the spec. Then I did a hard reset to clear the metadata so it could ingest the new offset range. For this kind of workflow we would need the executor to continue running even though the Supervisor is in COMPLETED state
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets = getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd() before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks() chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks() will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
+ */
+ private boolean isBoundedWorkComplete()
+ {
+ if (!ioConfig.isBounded()) {
+ return false;
+ }
+
+ // Check if task groups are empty (no tasks active or pending)
+ boolean noActiveTasks = activelyReadingTaskGroups.isEmpty();
+ boolean noPendingTasks = pendingCompletionTaskGroups.values().stream().allMatch(List::isEmpty);
+
+ if (!noActiveTasks || !noPendingTasks) {
+ return false;
+ }
+
+ // At this point, no tasks are running. Since createNewTasks() already ran,
+ // if tasks aren't running it means either:
+ // A) Tasks completed successfully and offset targets were reached (don't recreate)
+ // B) Tasks failed/killed and haven't reached targets (will recreate next run)
+ //
+ // To distinguish, we check if createNewTasks() would create new tasks.
+ // If hasTaskGroupReachedBoundedEnd() returns false for any group, createNewTasks()
+ // will create tasks next iteration, so we're not complete.
+ for (Integer groupId : partitionGroups.keySet()) {
+ if (!hasTaskGroupReachedBoundedEnd(groupId)) {
+ log.debug("TaskGroup[%d] has not reached bounded end, tasks will be recreated", groupId);
+ return false;
+ }
+ }
+
+ // All groups have reached their end offsets and no tasks are running.
+ // Work is complete!
+ log.info("All bounded tasks completed for supervisor[%s]", supervisorId);
+ return true;
+ }
+
+ /**
+ * Handle bounded processing completion by shutting down the supervisor.
+ * At this point, all task groups are already empty (verified by isBoundedWorkComplete),
+ * so we just need to mark the supervisor as completed.
+ */
+ private void handleBoundedCompletion()
+ {
+ log.info("Bounded processing complete for supervisor[%s]. Marking as COMPLETED.", supervisorId);
+ stateManager.maybeSetState(SupervisorStateManager.BasicState.COMPLETED);
Review Comment:
One workflow I was testing out was to submit a bounded Supervisor and have it run to completion. Then I adjusted the start/end offsets and re-submitted the spec. Then I did a hard reset to clear the metadata so it could ingest the new offset range. For this kind of workflow we would need the executor to continue running even after the initial completion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3165121359
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -249,6 +254,37 @@
int groupId,
ImmutableMap<PartitionIdType, SequenceOffsetType> startingSequences,
@Nullable ImmutableMap<PartitionIdType, SequenceOffsetType> unfilteredStartingSequencesForSequenceName,
+ @Nullable ImmutableMap<PartitionIdType, SequenceOffsetType> endSequences,
+ @Nullable DateTime minimumMessageTime,
+ @Nullable DateTime maximumMessageTime,
+ @Nullable Set<PartitionIdType> exclusiveStartSequenceNumberPartitions
+ )
+ {
+ this(
+ groupId,
+ startingSequences,
+ unfilteredStartingSequencesForSequenceName,
+ endSequences,
+ minimumMessageTime,
+ maximumMessageTime,
+ exclusiveStartSequenceNumberPartitions,
+ generateSequenceName(
+ unfilteredStartingSequencesForSequenceName == null
+ ? startingSequences
+ : unfilteredStartingSequencesForSequenceName,
+ minimumMessageTime,
+ maximumMessageTime,
+ spec.getDataSchema(),
+ taskTuningConfig
Review Comment:
## CodeQL / Deprecated method or constructor invocation
Invoking [SeekableStreamSupervisorSpec.getDataSchema](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/11164)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3174231540
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -259,6 +262,42 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
if (!this.getSource().equals(other.getSource())) {
throw InvalidInput.exception(ILLEGAL_INPUT_SOURCE_UPDATE_ERROR_MESSAGE, this.getSource(), other.getSource());
}
+
+ // Validate bounded stream configuration
+ validateBoundedStreamConfig(other);
+ }
+
+ /**
+ * Validates bounded stream configuration for the supervisor spec.
+ *
+ * @param spec the supervisor spec to validate
+ * @throws DruidException if the bounded stream configuration is invalid
+ */
+ protected void validateBoundedStreamConfig(SeekableStreamSupervisorSpec spec) throws DruidException
+ {
+ SeekableStreamSupervisorIOConfig ioConfig = spec.getIoConfig();
+
+ if (ioConfig.isBounded()) {
+ // Validate partition consistency
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ if (!boundedConfig.getStartSequenceNumbers().keySet().equals(boundedConfig.getEndSequenceNumbers().keySet())) {
+ throw InvalidInput.exception(
+ "Bounded stream config has mismatched partitions. Start: %s, End: %s",
+ boundedConfig.getStartSequenceNumbers().keySet(),
+ boundedConfig.getEndSequenceNumbers().keySet()
+ );
+ }
Review Comment:
I went ahead and removed this, along with the logging since these were largely the same validation checks that already run within BoundedStreamConfig
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3177423176
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java:
##########
@@ -0,0 +1,294 @@
+/*
+ * 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.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for bounded Kafka supervisors (one-time ingestion with explicit start/end offsets).
+ */
+public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
+{
+ private static final EmittingLogger log = new EmittingLogger(KafkaBoundedSupervisorTest.class);
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ @Test
+ public void test_boundedSupervisor_ingestsDataAndCompletes()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
+
+ // Get the current end offsets for all partitions
+ Map<String, Long> endOffsets = kafkaServer.getPartitionOffsets(topic);
+ Assertions.assertEquals(2, endOffsets.size(), "Should have 2 partitions");
+
+ // Create bounded config with start offset 0 and current end offsets
+ Map<String, Long> startOffsets = new HashMap<>();
+ startOffsets.put("0", 0L);
+ startOffsets.put("1", 0L);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for records to be ingested
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify row count
+ verifyRowCount(totalRecords);
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ Assertions.assertTrue(status.isHealthy());
+ }
+
+ @Test
+ public void test_boundedSupervisor_withEmptyRange_completesImmediately()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 1);
+
+ // Publish some records
+ publish1kRecords(topic, false);
+
+ // Get current offset
+ Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
+ Long currentOffset = currentOffsets.get("0");
+
+ // Create bounded config with start == end (empty range)
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("0", currentOffset);
+
+ BoundedStreamConfig boundedConfig = new BoundedStreamConfig(offsets, offsets);
+
+ // Create bounded supervisor
+ final KafkaSupervisorSpec supervisor = createBoundedKafkaSupervisor(
+ kafkaServer,
+ topic,
+ boundedConfig
+ );
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Wait for supervisor to transition to COMPLETED state
+ waitForSupervisorToComplete(supervisor.getId());
+
+ // Verify supervisor is in COMPLETED state
+ final SupervisorStatus status = cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("COMPLETED", status.getState());
+ }
+
+ private KafkaSupervisorSpec createBoundedKafkaSupervisor(
+ KafkaResource kafkaServer,
+ String topic,
+ BoundedStreamConfig boundedConfig
+ )
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null, null))
+ .withBoundedStreamConfig(boundedConfig)
+ )
+ .build(dataSource, topic);
+ }
+
+ @Test
+ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish records before creating supervisor
+ final int totalRecords = publish1kRecords(topic, false);
Review Comment:
## CodeQL / Unread local variable
Variable 'int totalRecords' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/11169)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3189385688
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4298,6 +4497,163 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if start >= end for all partitions (empty range)
+ // If so, there's no work to do - treat as already complete
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+ if (!isOffsetAtOrBeyond(start, end)) {
Review Comment:
[P1] Kinesis bounded ranges with start == end are skipped
The new empty-range check treats start >= end as completed for all bounded supervisors before creating any task. That is valid for Kafka's exclusive end offsets, but Kinesis declares bounded end offsets as inclusive and its task runner returns isEndOffsetExclusive() == false, so a Kinesis bounded ingestion for a single record where startSequenceNumbers equals endSequenceNumbers is marked COMPLETED without reading that record. This should be provider-aware, for example only treating equality as empty when the end offset is exclusive, while still rejecting/handling start > end appropriately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "aho135 (via GitHub)" <gi...@apache.org>.
aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3149723750
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java:
##########
@@ -78,7 +79,8 @@ public KafkaSupervisorIOConfig(
@JsonProperty("idleConfig") IdleConfig idleConfig,
@JsonProperty("stopTaskCount") Integer stopTaskCount,
@Nullable @JsonProperty("emitTimeLagMetrics") Boolean emitTimeLagMetrics,
- @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas
+ @Nullable @JsonProperty("serverPriorityToReplicas") Map<Integer, Integer> serverPriorityToReplicas,
+ @Nullable @JsonProperty("boundedStreamConfig") BoundedStreamConfig boundedStreamConfig
Review Comment:
Yup, the idea is for `taskDuration` and `completionTimeout` to work as they currently do. If a task exceeds the configured `taskDuration` but the offset range hasn't been fully processed yet then `isOffsetAtOrBeyond` will return false and the Supervisor will spin up a new task starting from the checkpointed metadata
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: Bounded Stream Supervisor (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3201843040
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4546,175 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end offsets.
+ * Used to determine if the task group completed successfully vs failed midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> startOffsets =
+ convertBoundedConfigMap(boundedConfig.getStartSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if all partitions have empty ranges
+ // For exclusive end offsets (Kafka): start >= end means empty
+ // For inclusive end offsets (Kinesis): only start > end means empty (start == end is one record)
+ boolean allPartitionsEmptyRange = true;
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType start = startOffsets.get(partition);
+ SequenceOffsetType end = endOffsets.get(partition);
+
+ boolean isEmpty;
+ if (isEndOffsetExclusive()) {
+ // Exclusive: empty if start >= end
+ isEmpty = isOffsetAtOrBeyond(start, end);
+ } else {
+ // Inclusive: empty only if start > end
+ isEmpty = isOffsetAtOrBeyond(start, end) && !start.equals(end);
+ }
+
+ if (!isEmpty) {
+ allPartitionsEmptyRange = false;
+ break;
+ }
+ }
+
+ if (allPartitionsEmptyRange) {
+ log.warn(
+ "TaskGroup[%d] has empty range for all partitions (start %s end). "
+ + "No work to do, marking as complete. Start: %s, End: %s",
+ groupId,
+ isEndOffsetExclusive() ? ">=" : ">",
+ startOffsets,
+ endOffsets
+ );
+ return true;
+ }
+
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets = getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
Review Comment:
[P1] Validate bounded metadata before completing from old offsets
A new bounded supervisor can silently complete using offsets from a previous bounded run with a different boundedStreamConfig. The mismatch check only runs in getOffsetFromStorageForPartition(), but isBoundedWorkComplete() calls hasTaskGroupReachedBoundedEnd() first, which reads raw metadata offsets and returns true when currentOffset >= the new end offset. For example, after a run [0,100] commits offset 100, posting [50,75] for the same datasource reaches COMPLETED without running tasks or throwing the documented metadata mismatch error. Validate the stored bounded config before treating metadata offsets as completion, or avoid using prior offsets when the bounded config differs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org