This is an automated email from the ASF dual-hosted git repository.
aho135 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new cf275580199 feat: resetOffsetsAndBackfill using bounded stream
supervisor (#19477)
cf275580199 is described below
commit cf275580199e93485062a7b1d7e1dbb4e2c6e2bf
Author: aho135 <[email protected]>
AuthorDate: Fri May 29 18:06:27 2026 -0700
feat: resetOffsetsAndBackfill using bounded stream supervisor (#19477)
* resetOffsetsAndBackfill using bounded stream supervisor
* Reject non-positive backfillTaskCount
* Reset supervisor after backfill Supervisor has already been started
* Add helper method specHasConcurrentLocks
* Fix doc reference
* Move validations into helper function
* Add embedded-test for resetSupervisorAndBackfill
* Remove flaky waitUntilPublishedRecordsAreIngested
* Update KafkaBoundedSupervisorTest.java
* Wait for supervisor to be RUNNING
* Use checkpointed offset if > requested reset offset to prevent duplicate
ingestion
* Update KafkaBoundedSupervisorTest.java
* Revert "Use checkpointed offset if > requested reset offset to prevent
duplicate ingestion"
resetOffsetsForwardOnly does not fully close the race it targets (the write
is
still unconditional) and the duplicate scenario it addresses is narrower
than
the overlap case, which cannot be solved without suspending the main
supervisor.
Accepting the limitation and documenting it is preferable to the added
complexity.
This reverts commit 89b5fec25e3a7dc88441d6e995564723557f2312.
* Doc update - duplication notice and Kinesis callout
* Rename endpoint from resetOffsetsAndBackfill to resetToLatestAndBackfill
* Update test name to reflect new endpoint
* Address clean up from review comments
* Log out start/end offsets
* Add abstract createBackfillSpec
* Unit test createBackfillSpec
* Fix deprecation notices
* Rename functions to align with new endpoint name
* Add null check and rename for consistency
---
docs/api-reference/supervisor-api.md | 103 ++++++++
.../indexing/KafkaBoundedSupervisorTest.java | 51 ++++
.../supervisor/RabbitStreamSupervisor.java | 6 +-
.../supervisor/RabbitStreamSupervisorSpec.java | 50 ++++
.../indexing/kafka/supervisor/KafkaSupervisor.java | 10 +-
.../kafka/supervisor/KafkaSupervisorSpec.java | 54 ++++
.../kafka/supervisor/KafkaSupervisorSpecTest.java | 33 +++
.../kinesis/supervisor/KinesisSupervisor.java | 4 +-
.../kinesis/supervisor/KinesisSupervisorSpec.java | 54 ++++
.../overlord/supervisor/SupervisorManager.java | 169 ++++++++++--
.../overlord/supervisor/SupervisorResource.java | 44 ++++
.../supervisor/SeekableStreamSupervisor.java | 10 +-
.../supervisor/SeekableStreamSupervisorSpec.java | 6 +
.../overlord/supervisor/SupervisorManagerTest.java | 290 +++++++++++++++++++++
.../supervisor/SupervisorResourceTest.java | 105 ++++++++
.../SeekableStreamSupervisorSpecTest.java | 10 +
.../SeekableStreamSupervisorStateTest.java | 8 +-
.../SeekableStreamSupervisorTestBase.java | 14 +-
.../indexing/MSQWorkerTaskLauncherRetryTest.java | 6 +
.../druid/rpc/indexing/NoopOverlordClient.java | 6 +
.../apache/druid/rpc/indexing/OverlordClient.java | 9 +
.../druid/rpc/indexing/OverlordClientImpl.java | 17 ++
.../testing/embedded/EmbeddedClusterApis.java | 10 +
23 files changed, 1021 insertions(+), 48 deletions(-)
diff --git a/docs/api-reference/supervisor-api.md
b/docs/api-reference/supervisor-api.md
index d321af14302..8f9c5c36dc5 100644
--- a/docs/api-reference/supervisor-api.md
+++ b/docs/api-reference/supervisor-api.md
@@ -3539,6 +3539,109 @@ when the supervisor's tasks restart, they resume
reading from `{"0": 100, "1": 1
```
</details>
+### Reset offsets to latest and start a backfill supervisor
+
+This endpoint is supported for Apache Kafka and RabbitMQ Stream supervisors.
Amazon Kinesis is not supported yet.
+
+Resets the supervisor to the latest available stream offsets and starts a new
bounded backfill supervisor to ingest the data in the skipped range.
+
+This endpoint is useful when a supervisor has fallen behind and you want to
catch it up to the latest offsets without losing the skipped data. The main
supervisor resumes ingesting from the latest offsets, while the backfill
supervisor processes the range from the previously checkpointed offsets up to
the latest offsets at the time of the reset.
+
+**Duplicate ingestion notice:** The main supervisor is not quiesced before the
reset. This means duplicate data can occur in two ways:
+- **Backfill overlap:** Any tasks that were in-flight at the time of the reset
may publish segments covering part of the backfill range before being shut down.
+- **Reset race:** If a task checkpoint is written to the metadata store
between when this endpoint captures the current offsets and when it applies the
reset, that checkpoint can be overwritten, causing the main supervisor to
re-ingest already-processed data.
+
+Both windows are narrow in practice, but cannot be fully eliminated without
manually suspending the main supervisor before calling this endpoint and
waiting for all pending tasks to complete.
+
+The following requirements must be met before calling this endpoint:
+
+- The supervisor must be a [streaming supervisor](../ingestion/supervisor.md).
+- The supervisor's `useEarliestSequenceNumber` property must be `false`.
+- The supervisor context must have `useConcurrentLocks` set to `true` to allow
the backfill supervisor's tasks to write concurrently with the main
supervisor's tasks.
+- The supervisor must be in a `RUNNING` state.
+
+The backfill supervisor has the same configuration as the source supervisor
except for its ID, which takes the form
`{supervisorId}_backfill_{randomSuffix}`, and its `boundedStreamConfig`, which
is set to the skipped offset range. If `backfillTaskCount` is specified, it
overrides the `taskCount` for the backfill supervisor only.
+
+#### URL
+
+`POST` `/druid/indexer/v1/supervisor/{supervisorId}/resetToLatestAndBackfill`
+
+#### Query parameters
+
+| Parameter | Type | Description | Default |
+|---------|---------|---------|---------|
+| `backfillTaskCount` | Integer | Number of parallel tasks for the backfill
supervisor. | Defaults to `taskCount` from the source supervisor if not
specified |
+
+#### Responses
+
+<Tabs>
+
+<TabItem value="5" label="200 SUCCESS">
+
+
+*Successfully reset and started backfill supervisor*
+
+</TabItem>
+<TabItem value="6" label="400 BAD REQUEST">
+
+
+*Supervisor does not meet requirements (wrong type,
`useEarliestSequenceNumber` is true, `useConcurrentLocks` not enabled, or
supervisor not RUNNING)*
+
+</TabItem>
+<TabItem value="7" label="404 NOT FOUND">
+
+
+*Invalid supervisor ID*
+
+</TabItem>
+<TabItem value="8" label="500 SERVER ERROR">
+
+
+*Failed to retrieve stream offsets or serialize the backfill spec*
+
+</TabItem>
+</Tabs>
+
+---
+
+#### Sample request
+
+The following example resets a supervisor named `social_media` and starts a
backfill supervisor with 2 tasks.
+
+<Tabs>
+
+<TabItem value="9" label="cURL">
+
+
+```shell
+curl --request POST
"http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/supervisor/social_media/resetToLatestAndBackfill?backfillTaskCount=2";
+```
+
+</TabItem>
+<TabItem value="10" label="HTTP">
+
+
+```HTTP
+POST
/druid/indexer/v1/supervisor/social_media/resetToLatestAndBackfill?backfillTaskCount=2
HTTP/1.1
+Host: http://ROUTER_IP:ROUTER_PORT
+```
+
+</TabItem>
+</Tabs>
+
+#### Sample response
+
+<details>
+ <summary>View the response</summary>
+
+ ```json
+{
+ "id": "social_media",
+ "backfillSupervisorId": "social_media_backfill_abcdefgh"
+}
+ ```
+</details>
+
### Terminate a supervisor
Terminates a supervisor and its associated indexing tasks, triggering the
publishing of their segments. When you terminate a supervisor, Druid places a
tombstone marker in the metadata store to prevent reloading on restart.
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
index 7e22d85d9ca..fa184418df5 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
@@ -292,6 +292,48 @@ public class KafkaBoundedSupervisorTest extends
StreamIndexTestBase
Assertions.assertEquals("UNHEALTHY_SUPERVISOR", status2.getState(),
"Supervisor state should be UNHEALTHY_SUPERVISOR");
}
+ @Test
+ public void test_resetToLatestAndBackfill()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Create a streaming supervisor with concurrent locks and
withUseEarliestSequenceNumber=false
+ final KafkaSupervisorSpec supervisor = createKafkaSupervisor(kafkaServer)
+ .withContext(Map.of("useConcurrentLocks", true))
+ .withIoConfig(io -> io
+ .withKafkaInputFormat(new JsonInputFormat(null, null, null, null,
null))
+ .withUseEarliestSequenceNumber(false)
+ )
+ .build(dataSource, topic);
+
+ cluster.callApi().postSupervisor(supervisor);
+
+ waitForSupervisorDetailedState(supervisor.getId(), "RUNNING");
+
+ final int totalRecords = publish1kRecords(topic, false);
+ waitUntilPublishedRecordsAreIngested(totalRecords);
+
+ // Reset the main supervisor and spin up a backfill supervisor.
+ // Since all records are already ingested before the call, the backfill
+ // supervisor will complete immediately without ingesting anything.
+ final Map<String, Object> result =
cluster.callApi().resetToLatestAndBackfill(supervisor.getId());
+ Assertions.assertEquals(supervisor.getId(), result.get("id"));
+ final String backfillSupervisorId = (String)
result.get("backfillSupervisorId");
+
+ // Wait for the backfill to finish
+ waitForSupervisorToComplete(backfillSupervisorId);
+
+ // Main supervisor should still be running
+ final SupervisorStatus mainStatus =
cluster.callApi().getSupervisorStatus(supervisor.getId());
+ Assertions.assertEquals("RUNNING", mainStatus.getState());
+ Assertions.assertTrue(mainStatus.isHealthy());
+
+ final SupervisorStatus backfillStatus =
cluster.callApi().getSupervisorStatus(backfillSupervisorId);
+ Assertions.assertEquals("COMPLETED", backfillStatus.getState());
+ Assertions.assertTrue(backfillStatus.isHealthy());
+ }
+
private void waitForSupervisorToComplete(String supervisorId)
{
overlord.latchableEmitter().waitForEvent(
@@ -301,6 +343,15 @@ public class KafkaBoundedSupervisorTest extends
StreamIndexTestBase
);
}
+ private void waitForSupervisorDetailedState(String supervisorId, String
detailedState)
+ {
+ overlord.latchableEmitter().waitForEvent(
+ event -> event.hasMetricName("supervisor/count")
+ .hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .hasDimension("detailedState", detailedState)
+ );
+ }
+
private void waitForSupervisorToBeUnhealthy(String supervisorId)
{
overlord.latchableEmitter().waitForEvent(
diff --git
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java
index 6099105b337..04973a5272f 100644
---
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java
+++
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java
@@ -322,7 +322,7 @@ public class RabbitStreamSupervisor extends
SeekableStreamSupervisor<String, Lon
}
@Override
- protected RabbitStreamDataSourceMetadata
createDataSourceMetaDataForReset(String topic, Map<String, Long> map)
+ public RabbitStreamDataSourceMetadata
createDataSourceMetaDataForReset(String topic, Map<String, Long> map)
{
return new RabbitStreamDataSourceMetadata(new
SeekableStreamEndSequenceNumbers<>(topic, map));
}
@@ -408,7 +408,7 @@ public class RabbitStreamSupervisor extends
SeekableStreamSupervisor<String, Lon
}
@Override
- protected void updatePartitionLagFromStream()
+ public void updatePartitionLagFromStream()
{
getRecordSupplierLock().lock();
@@ -435,7 +435,7 @@ public class RabbitStreamSupervisor extends
SeekableStreamSupervisor<String, Lon
}
@Override
- protected Map<String, Long> getLatestSequencesFromStream()
+ public Map<String, Long> getLatestSequencesFromStream()
{
return latestSequenceFromStream != null ? latestSequenceFromStream : new
HashMap<>();
}
diff --git
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
index 4a445f6f1c1..4763a949a61 100644
---
a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
+++
b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
@@ -30,6 +30,7 @@ import org.apache.druid.indexing.overlord.TaskStorage;
import org.apache.druid.indexing.overlord.supervisor.Supervisor;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
import
org.apache.druid.indexing.rabbitstream.RabbitStreamIndexTaskClientFactory;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.metrics.DruidMonitorSchedulerConfig;
@@ -155,6 +156,55 @@ public class RabbitStreamSupervisorSpec extends
SeekableStreamSupervisorSpec
supervisorStateManagerConfig);
}
+ @Override
+ public RabbitStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ RabbitStreamSupervisorIOConfig ioConfig = getSpec().getIOConfig();
+ RabbitStreamSupervisorIOConfig backfillIoConfig = new
RabbitStreamSupervisorIOConfig(
+ ioConfig.getStream(),
+ ioConfig.getUri(),
+ ioConfig.getInputFormat(),
+ ioConfig.getReplicas(),
+ taskCount != null ? taskCount : ioConfig.getTaskCount(),
+ ioConfig.getTaskDuration().toPeriod(),
+ ioConfig.getConsumerProperties(),
+ ioConfig.getAutoScalerConfig(),
+ ioConfig.getPollTimeout(),
+ ioConfig.getStartDelay().toPeriod(),
+ ioConfig.getPeriod().toPeriod(),
+ ioConfig.getCompletionTimeout().toPeriod(),
+ ioConfig.isUseEarliestSequenceNumber(),
+ ioConfig.getLateMessageRejectionPeriod().isPresent() ?
ioConfig.getLateMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getEarlyMessageRejectionPeriod().isPresent() ?
ioConfig.getEarlyMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getLateMessageRejectionStartDateTime().isPresent() ?
ioConfig.getLateMessageRejectionStartDateTime().get() : null,
+ ioConfig.getStopTaskCount(),
+ ioConfig.getServerPriorityToReplicas(),
+ boundedStreamConfig
+ );
+ return new RabbitStreamSupervisorSpec(
+ backfillId,
+ null,
+ getSpec().getDataSchema(),
+ getSpec().getTuningConfig(),
+ backfillIoConfig,
+ getContext(),
+ isSuspended(),
+ taskStorage,
+ taskMaster,
+ indexerMetadataStorageCoordinator,
+ (RabbitStreamIndexTaskClientFactory) indexTaskClientFactory,
+ mapper,
+ emitter,
+ monitorSchedulerConfig,
+ rowIngestionMetersFactory,
+ supervisorStateManagerConfig
+ );
+ }
+
@Override
public String toString()
{
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
index 727eb52db27..5863284cc2d 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
@@ -356,7 +356,7 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
}
@Override
- protected KafkaDataSourceMetadata createDataSourceMetaDataForReset(String
topic, Map<KafkaTopicPartition, Long> map)
+ public KafkaDataSourceMetadata createDataSourceMetaDataForReset(String
topic, Map<KafkaTopicPartition, Long> map)
{
return new KafkaDataSourceMetadata(new
SeekableStreamEndSequenceNumbers<>(topic, map));
}
@@ -548,7 +548,7 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
* </p>
*/
@Override
- protected void updatePartitionLagFromStream()
+ public void updatePartitionLagFromStream()
{
if (getIoConfig().isEmitTimeLagMetrics()) {
updatePartitionTimeAndRecordLagFromStream();
@@ -597,7 +597,7 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
}
@Override
- protected Map<KafkaTopicPartition, Long> getLatestSequencesFromStream()
+ public Map<KafkaTopicPartition, Long> getLatestSequencesFromStream()
{
return offsetSnapshotRef.get().getLatestOffsetsFromStream();
}
@@ -630,7 +630,7 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
* Gets the offsets as stored in the metadata store. The map returned will
only contain
* offsets from topic partitions that match the current supervisor config
stream. This
* override is needed because in the case of multi-topic, a user could have
updated the supervisor
- * config from single topic to mult-topic, where the new multi-topic pattern
regex matches the
+ * config from single topic to multi-topic, where the new multi-topic
pattern regex matches the
* old config single topic. Without this override, the previously stored
metadata for the single
* topic would be deemed as different from the currently configure stream,
and not be included in
* the offset map returned. This implementation handles these cases
appropriately.
@@ -640,7 +640,7 @@ public class KafkaSupervisor extends
SeekableStreamSupervisor<KafkaTopicPartitio
* updated to single topic or multi-topic depending on the supervisor
config, as needed.
*/
@Override
- protected Map<KafkaTopicPartition, Long> getOffsetsFromMetadataStorage()
+ public Map<KafkaTopicPartition, Long> getOffsetsFromMetadataStorage()
{
final DataSourceMetadata dataSourceMetadata = retrieveDataSourceMetadata();
if (checkSourceMetadataMatch(dataSourceMetadata)) {
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
index b607ade1acf..31d3e8fad69 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
@@ -36,6 +36,7 @@ import org.apache.druid.indexing.overlord.TaskStorage;
import org.apache.druid.indexing.overlord.supervisor.Supervisor;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
@@ -173,6 +174,59 @@ public class KafkaSupervisorSpec extends
SeekableStreamSupervisorSpec
);
}
+ @Override
+ public KafkaSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ KafkaSupervisorIOConfig ioConfig = getSpec().getIOConfig();
+ KafkaSupervisorIOConfig backfillIoConfig = new KafkaSupervisorIOConfig(
+ ioConfig.getTopic(),
+ ioConfig.getTopicPattern(),
+ ioConfig.getInputFormat(),
+ ioConfig.getReplicas(),
+ taskCount != null ? taskCount : ioConfig.getTaskCount(),
+ ioConfig.getTaskDuration().toPeriod(),
+ ioConfig.getConsumerProperties(),
+ ioConfig.getAutoScalerConfig(),
+ ioConfig.getLagAggregator(),
+ ioConfig.getPollTimeout(),
+ ioConfig.getStartDelay().toPeriod(),
+ ioConfig.getPeriod().toPeriod(),
+ ioConfig.isUseEarliestSequenceNumber(),
+ ioConfig.getCompletionTimeout().toPeriod(),
+ ioConfig.getLateMessageRejectionPeriod().isPresent() ?
ioConfig.getLateMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getEarlyMessageRejectionPeriod().isPresent() ?
ioConfig.getEarlyMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getLateMessageRejectionStartDateTime().isPresent() ?
ioConfig.getLateMessageRejectionStartDateTime().get() : null,
+ ioConfig.getConfigOverrides(),
+ ioConfig.getIdleConfig(),
+ ioConfig.getStopTaskCount(),
+ ioConfig.isEmitTimeLagMetrics(),
+ ioConfig.getServerPriorityToReplicas(),
+ boundedStreamConfig
+ );
+ return new KafkaSupervisorSpec(
+ backfillId,
+ null,
+ getSpec().getDataSchema(),
+ getSpec().getTuningConfig(),
+ backfillIoConfig,
+ getContext(),
+ isSuspended(),
+ taskStorage,
+ taskMaster,
+ indexerMetadataStorageCoordinator,
+ (KafkaIndexTaskClientFactory) indexTaskClientFactory,
+ mapper,
+ emitter,
+ monitorSchedulerConfig,
+ rowIngestionMetersFactory,
+ supervisorStateManagerConfig
+ );
+ }
+
/**
* Extends {@link SeekableStreamSupervisorSpec#validateSpecUpdateTo} to
ensure that the proposed spec and current spec are either both multi-topic or
both single-topic.
* <p>
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpecTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpecTest.java
index 8879ff6d975..06ca9b64ced 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpecTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpecTest.java
@@ -32,6 +32,7 @@ import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskStorage;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
import
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
@@ -564,6 +565,38 @@ public class KafkaSupervisorSpecTest
sourceSpec.validateSpecUpdateTo(validDestSpec);
}
+ @Test
+ public void testCreateBackfillSpec()
+ {
+ KafkaSupervisorSpec spec = new KafkaSupervisorSpecBuilder()
+ .withDataSchema(
+ schema -> schema
+ .withTimestamp(TimestampSpec.DEFAULT)
+ .withAggregators(new CountAggregatorFactory("rows"))
+ .withGranularity(new
UniformGranularitySpec(Granularities.HOUR, Granularities.NONE, null))
+ )
+ .withIoConfig(
+ ioConfig -> ioConfig
+ .withJsonInputFormat()
+ .withConsumerProperties(Map.of("bootstrap.servers",
"localhost:9092"))
+ .withTaskCount(3)
+ )
+ .build("testDs", "metrics");
+
+ BoundedStreamConfig boundedStreamConfig = new BoundedStreamConfig(
+ Map.of("0", 100L, "1", 200L),
+ Map.of("0", 500L, "1", 600L)
+ );
+
+ KafkaSupervisorSpec backfill = (KafkaSupervisorSpec)
spec.createBackfillSpec("backfill-id", boundedStreamConfig, 2);
+
+ Assert.assertEquals("backfill-id", backfill.getId());
+ Assert.assertEquals("testDs",
backfill.getSpec().getDataSchema().getDataSource());
+ Assert.assertEquals("metrics",
backfill.getSpec().getIOConfig().getTopic());
+ Assert.assertEquals(2, backfill.getSpec().getIOConfig().getTaskCount());
+ Assert.assertEquals(boundedStreamConfig,
backfill.getSpec().getIOConfig().getBoundedStreamConfig());
+ }
+
private KafkaSupervisorSpec getSpec(String topic, String topicPattern)
{
KafkaSupervisorSpecBuilder builder = new KafkaSupervisorSpecBuilder()
diff --git
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
index 0f91fc0965d..3f1f4034f3c 100644
---
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
+++
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
@@ -321,7 +321,7 @@ public class KinesisSupervisor extends
SeekableStreamSupervisor<String, String,
}
@Override
- protected SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
+ public SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
String stream,
Map<String, String> map
)
@@ -336,7 +336,7 @@ public class KinesisSupervisor extends
SeekableStreamSupervisor<String, String,
}
@Override
- protected void updatePartitionLagFromStream()
+ public void updatePartitionLagFromStream()
{
KinesisRecordSupplier supplier = (KinesisRecordSupplier) recordSupplier;
// this recordSupplier method is thread safe, so does not need to acquire
the recordSupplierLock
diff --git
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
index 8e661571680..4899337797b 100644
---
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
+++
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
@@ -35,6 +35,7 @@ import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskStorage;
import org.apache.druid.indexing.overlord.supervisor.Supervisor;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.metrics.DruidMonitorSchedulerConfig;
@@ -193,4 +194,57 @@ public class KinesisSupervisorSpec extends
SeekableStreamSupervisorSpec
supervisorStateManagerConfig
);
}
+
+ @Override
+ public KinesisSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ KinesisSupervisorIOConfig ioConfig = getSpec().getIOConfig();
+ KinesisSupervisorIOConfig backfillIoConfig = new KinesisSupervisorIOConfig(
+ ioConfig.getStream(),
+ ioConfig.getInputFormat(),
+ ioConfig.getEndpoint(),
+ null,
+ ioConfig.getReplicas(),
+ taskCount != null ? taskCount : ioConfig.getTaskCount(),
+ ioConfig.getTaskDuration().toPeriod(),
+ ioConfig.getStartDelay().toPeriod(),
+ ioConfig.getPeriod().toPeriod(),
+ ioConfig.isUseEarliestSequenceNumber(),
+ ioConfig.getCompletionTimeout().toPeriod(),
+ ioConfig.getLateMessageRejectionPeriod().isPresent() ?
ioConfig.getLateMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getEarlyMessageRejectionPeriod().isPresent() ?
ioConfig.getEarlyMessageRejectionPeriod().get().toPeriod() : null,
+ ioConfig.getLateMessageRejectionStartDateTime().isPresent() ?
ioConfig.getLateMessageRejectionStartDateTime().get() : null,
+ ioConfig.getRecordsPerFetch(),
+ ioConfig.getFetchDelayMillis(),
+ ioConfig.getAwsAssumedRoleArn(),
+ ioConfig.getAwsExternalId(),
+ ioConfig.getAutoScalerConfig(),
+ ioConfig.isDeaggregate(),
+ ioConfig.getServerPriorityToReplicas(),
+ boundedStreamConfig
+ );
+ return new KinesisSupervisorSpec(
+ backfillId,
+ null,
+ getSpec().getDataSchema(),
+ getSpec().getTuningConfig(),
+ backfillIoConfig,
+ getContext(),
+ isSuspended(),
+ taskStorage,
+ taskMaster,
+ indexerMetadataStorageCoordinator,
+ (KinesisIndexTaskClientFactory) indexTaskClientFactory,
+ mapper,
+ emitter,
+ monitorSchedulerConfig,
+ rowIngestionMetersFactory,
+ awsCredentialsConfig,
+ supervisorStateManagerConfig
+ );
+ }
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
index 52f3cba7fc1..fa7d96634ae 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
@@ -23,9 +23,11 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.inject.Inject;
import org.apache.druid.common.guava.FutureUtils;
+import org.apache.druid.common.utils.IdUtils;
import org.apache.druid.error.DruidException;
import org.apache.druid.error.InvalidInput;
import org.apache.druid.error.NotFound;
@@ -35,8 +37,11 @@ import org.apache.druid.indexing.common.task.Tasks;
import org.apache.druid.indexing.overlord.DataSourceMetadata;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
import
org.apache.druid.indexing.seekablestream.SeekableStreamDataSourceMetadata;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
@@ -129,33 +134,8 @@ public class SupervisorManager implements
SupervisorStatsProvider
final Supervisor supervisor = entry.getValue().lhs;
final SupervisorSpec supervisorSpec = entry.getValue().rhs;
- boolean hasAppendLock = Tasks.DEFAULT_USE_CONCURRENT_LOCKS;
- if (supervisorSpec instanceof SeekableStreamSupervisorSpec) {
- SeekableStreamSupervisorSpec seekableStreamSupervisorSpec =
(SeekableStreamSupervisorSpec) supervisorSpec;
- Map<String, Object> context =
seekableStreamSupervisorSpec.getContext();
- if (context != null) {
- Boolean useConcurrentLocks = QueryContexts.getAsBoolean(
- Tasks.USE_CONCURRENT_LOCKS,
- context.get(Tasks.USE_CONCURRENT_LOCKS)
- );
- if (useConcurrentLocks == null) {
- TaskLockType taskLockType = QueryContexts.getAsEnum(
- Tasks.TASK_LOCK_TYPE,
- context.get(Tasks.TASK_LOCK_TYPE),
- TaskLockType.class
- );
- if (taskLockType == null) {
- hasAppendLock = Tasks.DEFAULT_USE_CONCURRENT_LOCKS;
- } else if (taskLockType == TaskLockType.APPEND) {
- hasAppendLock = true;
- } else {
- hasAppendLock = false;
- }
- } else {
- hasAppendLock = useConcurrentLocks;
- }
- }
- }
+ boolean hasAppendLock = supervisorSpec instanceof
SeekableStreamSupervisorSpec
+ &&
specHasConcurrentLocks((SeekableStreamSupervisorSpec) supervisorSpec);
if (supervisor instanceof SeekableStreamSupervisor
&& !supervisorSpec.isSuspended()
@@ -393,6 +373,116 @@ public class SupervisorManager implements
SupervisorStatsProvider
return true;
}
+ /**
+ * Resets a supervisor to the latest stream offsets and starts a bounded
backfill supervisor to
+ * process the skipped range from the previously checkpointed offsets up to
the latest offsets.
+ *
+ * @param id supervisor ID
+ * @param backfillTaskCount number of tasks for the backfill supervisor, or
null to inherit from the source spec
+ * @return map with {@code "id"} (the original supervisor ID) and {@code
"backfillSupervisorId"}
+ * @throws IllegalArgumentException if the supervisor is not a {@link
SeekableStreamSupervisor},
+ * if {@code useEarliestSequenceNumber} is
true,
+ * if {@code useConcurrentLocks} is not set
to true in the supervisor context,
+ * or if the supervisor is not in a RUNNING
state
+ * @throws IllegalStateException if the latest or checkpointed offsets
cannot be retrieved,
+ * or if the backfill spec cannot be
serialized
+ */
+ public Map<String, Object> resetToLatestAndBackfill(String id, @Nullable
Integer backfillTaskCount)
+ {
+ Preconditions.checkState(started, "SupervisorManager not started");
+ Preconditions.checkNotNull(id, "id");
+
+ Pair<Supervisor, SupervisorSpec> supervisor = supervisors.get(id);
+
+ if (supervisor == null) {
+ throw new IAE("Supervisor[%s] does not exist", id);
+ }
+
+ if (!(supervisor.lhs instanceof SeekableStreamSupervisor)) {
+ throw new IAE("Supervisor[%s] is not a streaming supervisor", id);
+ }
+
+ SeekableStreamSupervisor streamSupervisor = (SeekableStreamSupervisor)
supervisor.lhs;
+ SeekableStreamSupervisorSpec streamSpec = (SeekableStreamSupervisorSpec)
supervisor.rhs;
+
+ validateResetAndBackfill(id, streamSupervisor, streamSpec);
+
+ log.info("Capturing latest offsets from stream for supervisor[%s]", id);
+ streamSupervisor.updatePartitionLagFromStream();
+ Map<?, ?> endOffsets = streamSupervisor.getLatestSequencesFromStream();
+
+ log.info("Capturing checkpointed offsets for supervisor[%s]", id);
+ Map<?, ?> startOffsets = streamSupervisor.getOffsetsFromMetadataStorage();
+
+ if (endOffsets == null || endOffsets.isEmpty()) {
+ throw new ISE("Skipping reset: Failed to get latest offsets from stream
for supervisor[%s]", id);
+ }
+ if (startOffsets == null || startOffsets.isEmpty()) {
+ throw new ISE("Skipping reset: Failed to get checkpointed offsets for
supervisor[%s]", id);
+ }
+
+ String backfillSupervisorId = IdUtils.getRandomIdWithPrefix(id +
"_backfill");
+
+ try {
+ Map<String, Object> normalizedStartOffsets =
jsonMapper.readValue(jsonMapper.writeValueAsString(startOffsets), Map.class);
+ Map<String, Object> normalizedEndOffsets =
jsonMapper.readValue(jsonMapper.writeValueAsString(endOffsets), Map.class);
+ BoundedStreamConfig boundedStreamConfig = new
BoundedStreamConfig(normalizedStartOffsets, normalizedEndOffsets);
+ SupervisorSpec backfillSpec =
streamSpec.createBackfillSpec(backfillSupervisorId, boundedStreamConfig,
backfillTaskCount);
+ createOrUpdateAndStartSupervisor(backfillSpec);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE(e, "Failed to serialize offsets for backfill
supervisor[%s]", backfillSupervisorId);
+ }
+
+ log.info(
+ "Started backfill supervisor[%s] for supervisor[%s] with
startOffsets[%s] and endOffsets[%s]",
+ backfillSupervisorId,
+ id,
+ startOffsets,
+ endOffsets
+ );
+
+ log.info("Resetting supervisor[%s] metadata to latest offsets", id);
+ DataSourceMetadata resetMetadata =
streamSupervisor.createDataSourceMetaDataForReset(
+ streamSupervisor.getIoConfig().getStream(),
+ endOffsets
+ );
+
+ streamSupervisor.resetOffsets(resetMetadata);
+
+ // Reset autoscaler if present
+ SupervisorTaskAutoScaler autoscaler = autoscalers.get(id);
+ if (autoscaler != null) {
+ autoscaler.reset();
+ }
+
+ return ImmutableMap.of(
+ "id", id,
+ "backfillSupervisorId", backfillSupervisorId
+ );
+ }
+
+ private void validateResetAndBackfill(
+ String id,
+ SeekableStreamSupervisor streamSupervisor,
+ SeekableStreamSupervisorSpec streamSpec
+ )
+ {
+ if (streamSupervisor.getIoConfig().isUseEarliestSequenceNumber()) {
+ throw new IAE("Reset with skipped offsets is not supported when
useEarliestOffset is true.");
+ }
+
+ if (!specHasConcurrentLocks(streamSpec)) {
+ throw new IAE(
+ "Backfill tasks require 'useConcurrentLocks' to be set to true in
the supervisor context to allow concurrent writes with the main supervisor
tasks"
+ );
+ }
+
+ if (streamSupervisor.getState() !=
SupervisorStateManager.BasicState.RUNNING) {
+ throw new IAE("Supervisor[%s] must be in a RUNNING state to perform a
reset and backfill", id);
+ }
+ }
+
public boolean checkPointDataSourceMetadata(
String supervisorId,
int taskGroupId,
@@ -631,4 +721,29 @@ public class SupervisorManager implements
SupervisorStatsProvider
return supervisor == null ? null : supervisor.rhs;
}
}
+
+ /**
+ * Returns true if the spec's context enables concurrent (append) locks,
accepting both
+ * {@code useConcurrentLocks: true} (or any truthy string) and {@code
taskLockType: APPEND}.
+ */
+ private static boolean specHasConcurrentLocks(SeekableStreamSupervisorSpec
spec)
+ {
+ Map<String, Object> context = spec.getContext();
+ if (context == null) {
+ return Tasks.DEFAULT_USE_CONCURRENT_LOCKS;
+ }
+ Boolean useConcurrentLocks = QueryContexts.getAsBoolean(
+ Tasks.USE_CONCURRENT_LOCKS,
+ context.get(Tasks.USE_CONCURRENT_LOCKS)
+ );
+ if (useConcurrentLocks != null) {
+ return useConcurrentLocks;
+ }
+ TaskLockType taskLockType = QueryContexts.getAsEnum(
+ Tasks.TASK_LOCK_TYPE,
+ context.get(Tasks.TASK_LOCK_TYPE),
+ TaskLockType.class
+ );
+ return taskLockType == TaskLockType.APPEND;
+ }
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
index aff9edf19af..8d0e04eb798 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
@@ -640,6 +640,50 @@ public class SupervisorResource
);
}
+ @POST
+ @Path("/{id}/resetToLatestAndBackfill")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ResourceFilters(SupervisorResourceFilter.class)
+ public Response resetToLatestAndBackfill(
+ @PathParam("id") final String id,
+ @QueryParam("backfillTaskCount") @Nullable final Integer
backfillTaskCount
+ )
+ {
+ return handleResetToLatestAndBackfill(id, backfillTaskCount);
+ }
+
+ private Response handleResetToLatestAndBackfill(final String id, @Nullable
final Integer backfillTaskCount)
+ {
+ if (backfillTaskCount != null && backfillTaskCount < 1) {
+ return Response.status(Response.Status.BAD_REQUEST)
+ .entity(ImmutableMap.of("error", "backfillTaskCount must
be a positive integer"))
+ .build();
+ }
+ return asLeaderWithSupervisorManager(
+ manager -> {
+ if (!manager.getSupervisorIds().contains(id)) {
+ return Response.status(Response.Status.NOT_FOUND)
+ .entity(ImmutableMap.of("error",
StringUtils.format("[%s] does not exist", id)))
+ .build();
+ }
+ try {
+ Map<String, Object> result = manager.resetToLatestAndBackfill(id,
backfillTaskCount);
+ return Response.ok(result).build();
+ }
+ catch (IllegalArgumentException e) {
+ return Response.status(Response.Status.BAD_REQUEST)
+ .entity(ImmutableMap.of("error", e.getMessage()))
+ .build();
+ }
+ catch (Exception e) {
+ return Response.serverError()
+ .entity(ImmutableMap.of("error", e.getMessage()))
+ .build();
+ }
+ }
+ );
+ }
+
private Response asLeaderWithSupervisorManager(Function<SupervisorManager,
Response> f)
{
Optional<SupervisorManager> supervisorManager =
taskMaster.getSupervisorManager();
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index 9bef543496b..74329c68e1d 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -2195,7 +2195,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
final boolean metadataUpdateSuccess;
final DataSourceMetadata metadata =
indexerMetadataStorageCoordinator.retrieveDataSourceMetadata(supervisorId);
if (metadata == null) {
- log.info("Checkpointed metadata in null for supervisor[%s] for
dataSource[%s] - inserting metadata[%s]", supervisorId, dataSource,
resetMetadata);
+ log.info("Checkpointed metadata is null for supervisor[%s] for
dataSource[%s] - inserting metadata[%s]", supervisorId, dataSource,
resetMetadata);
metadataUpdateSuccess =
indexerMetadataStorageCoordinator.insertDataSourceMetadata(supervisorId,
resetMetadata);
} else {
if (!checkSourceMetadataMatch(metadata)) {
@@ -3311,7 +3311,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
/**
* gets mapping of partitions in stream to their latest offsets.
*/
- protected Map<PartitionIdType, SequenceOffsetType>
getLatestSequencesFromStream()
+ public Map<PartitionIdType, SequenceOffsetType>
getLatestSequencesFromStream()
{
return new HashMap<>();
}
@@ -4589,7 +4589,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
}
- protected Map<PartitionIdType, SequenceOffsetType>
getOffsetsFromMetadataStorage()
+ public Map<PartitionIdType, SequenceOffsetType>
getOffsetsFromMetadataStorage()
{
final DataSourceMetadata dataSourceMetadata = retrieveDataSourceMetadata();
if (dataSourceMetadata instanceof SeekableStreamDataSourceMetadata
@@ -4976,7 +4976,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
coalesceAndAwait(futures);
}
- protected abstract void updatePartitionLagFromStream();
+ public abstract void updatePartitionLagFromStream();
/**
* Gets 'lag' of currently processed offset behind latest offset as a
measure of difference between offsets.
@@ -5233,7 +5233,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
* @param map partitionId -> sequence
* @return specific instance of datasource metadata
*/
- protected abstract SeekableStreamDataSourceMetadata<PartitionIdType,
SequenceOffsetType> createDataSourceMetaDataForReset(
+ public abstract SeekableStreamDataSourceMetadata<PartitionIdType,
SequenceOffsetType> createDataSourceMetaDataForReset(
String stream,
Map<PartitionIdType, SequenceOffsetType> map
);
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
index 842f0de4774..ecbd51757c3 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
@@ -297,4 +297,10 @@ public abstract class SeekableStreamSupervisorSpec
implements SupervisorSpec
protected abstract SeekableStreamSupervisorSpec toggleSuspend(boolean
suspend);
+ public abstract SeekableStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ );
+
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
index 525444e23de..199e004b424 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
@@ -19,6 +19,9 @@
package org.apache.druid.indexing.overlord.supervisor;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
@@ -26,6 +29,8 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.druid.data.input.impl.ByteEntity;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.error.DruidException;
import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.error.InvalidInput;
@@ -35,7 +40,11 @@ import org.apache.druid.indexing.overlord.DataSourceMetadata;
import org.apache.druid.indexing.overlord.ObjectMetadata;
import
org.apache.druid.indexing.seekablestream.SeekableStreamStartSequenceNumbers;
import
org.apache.druid.indexing.seekablestream.TestSeekableStreamDataSourceMetadata;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor;
+import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
+import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
@@ -43,6 +52,7 @@ import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.metadata.MetadataSupervisorManager;
import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.server.metrics.SupervisorStatsProvider;
import org.apache.druid.timeline.partition.NumberedShardSpec;
@@ -59,6 +69,7 @@ import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
+import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.util.Collection;
import java.util.Collections;
@@ -1068,6 +1079,186 @@ public class SupervisorManagerTest extends
EasyMockSupport
);
}
+ @Test
+ public void testResetToLatestAndBackfill() throws Exception
+ {
+
EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(ImmutableMap.of());
+ replayAll();
+ manager.start();
+
+ final ConcurrentHashMap<String, Pair<Supervisor, SupervisorSpec>>
supervisorsMap = getSupervisorsMap();
+ final SeekableStreamSupervisorSpec streamSpec =
EasyMock.createNiceMock(SeekableStreamSupervisorSpec.class);
+ final SeekableStreamSupervisor streamSupervisor =
EasyMock.createNiceMock(SeekableStreamSupervisor.class);
+ final SeekableStreamSupervisorIOConfig ioConfig =
EasyMock.createNiceMock(SeekableStreamSupervisorIOConfig.class);
+
+ // non-SeekableStream supervisor → IAE
+ // Use a concrete anonymous Supervisor (not a mock) to reliably fail
instanceof SeekableStreamSupervisor
+ final Supervisor nonStreamSupervisor = new Supervisor()
+ {
+ @Override
+ public void start()
+ {
+ }
+
+ @Override
+ public void stop(boolean stopGracefully)
+ {
+ }
+
+ @Override
+ public SupervisorReport getStatus()
+ {
+ return null;
+ }
+
+ @Override
+ public SupervisorStateManager.State getState()
+ {
+ return null;
+ }
+
+ @Override
+ public void reset(DataSourceMetadata dataSourceMetadata)
+ {
+ }
+ };
+ supervisorsMap.put("id1", Pair.of(nonStreamSupervisor, streamSpec));
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+
+ // useEarliestSequenceNumber=true → IAE
+ supervisorsMap.put("id1", Pair.of(streamSupervisor, streamSpec));
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(true).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // useConcurrentLocks not set (null context) → IAE
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+ EasyMock.expect(streamSpec.getContext()).andReturn(null).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // useConcurrentLocks=false → IAE
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("useConcurrentLocks",
false)).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // useConcurrentLocks="true" (string) → accepted, fails at next guard (not
RUNNING)
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("useConcurrentLocks",
"true")).once();
+
EasyMock.expect(streamSupervisor.getState()).andReturn(SupervisorStateManager.BasicState.SUSPENDED).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // taskLockType=APPEND → accepted, fails at next guard (not RUNNING)
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("taskLockType",
"APPEND")).once();
+
EasyMock.expect(streamSupervisor.getState()).andReturn(SupervisorStateManager.BasicState.SUSPENDED).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // supervisor not RUNNING → IAE
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("useConcurrentLocks",
true)).once();
+
EasyMock.expect(streamSupervisor.getState()).andReturn(SupervisorStateManager.BasicState.SUSPENDED).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // empty latest offsets → ISE
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("useConcurrentLocks",
true)).once();
+
EasyMock.expect(streamSupervisor.getState()).andReturn(SupervisorStateManager.BasicState.RUNNING).once();
+ streamSupervisor.updatePartitionLagFromStream();
+ EasyMock.expectLastCall().once();
+
EasyMock.expect(streamSupervisor.getLatestSequencesFromStream()).andReturn(ImmutableMap.of()).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalStateException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+ EasyMock.reset(streamSupervisor, streamSpec, ioConfig);
+
+ // empty start offsets from metadata → ISE
+
EasyMock.expect(streamSupervisor.getIoConfig()).andReturn(ioConfig).anyTimes();
+
EasyMock.expect(ioConfig.isUseEarliestSequenceNumber()).andReturn(false).once();
+
EasyMock.expect(streamSpec.getContext()).andReturn(ImmutableMap.of("useConcurrentLocks",
true)).once();
+
EasyMock.expect(streamSupervisor.getState()).andReturn(SupervisorStateManager.BasicState.RUNNING).once();
+ streamSupervisor.updatePartitionLagFromStream();
+ EasyMock.expectLastCall().once();
+
EasyMock.expect(streamSupervisor.getLatestSequencesFromStream()).andReturn(ImmutableMap.of("0",
100L)).once();
+
EasyMock.expect(streamSupervisor.getOffsetsFromMetadataStorage()).andReturn(ImmutableMap.of()).once();
+ EasyMock.replay(streamSupervisor, streamSpec, ioConfig);
+ Assert.assertThrows(
+ IllegalStateException.class,
+ () -> manager.resetToLatestAndBackfill("id1", null)
+ );
+
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateBackfillSpec()
+ {
+ final TestBackfillSupervisorSpec.IOConfig ioConfig = new
TestBackfillSupervisorSpec.IOConfig("test-stream", null, null);
+ final TestBackfillSupervisorSpec.IngestionSpec ingestionSpec = new
TestBackfillSupervisorSpec.IngestionSpec(ioConfig);
+ final SeekableStreamSupervisorSpec sourceSpec = new
TestBackfillSupervisorSpec("original-id", ingestionSpec);
+
+ final BoundedStreamConfig boundedStreamConfig = new BoundedStreamConfig(
+ ImmutableMap.of("0", 100L),
+ ImmutableMap.of("0", 200L)
+ );
+
+ // Without overriding taskCount
+ final SupervisorSpec backfillSpec =
sourceSpec.createBackfillSpec("backfill-id", boundedStreamConfig, null);
+ Assert.assertEquals("backfill-id", backfillSpec.getId());
+ final TestBackfillSupervisorSpec backfillCast =
(TestBackfillSupervisorSpec) backfillSpec;
+ final BoundedStreamConfig actualConfig =
backfillCast.getIoConfig().getBoundedStreamConfig();
+ Assert.assertNotNull(actualConfig);
+ Assert.assertEquals(ImmutableMap.of("0", 100L),
actualConfig.getStartSequenceNumbers());
+ Assert.assertEquals(ImmutableMap.of("0", 200L),
actualConfig.getEndSequenceNumbers());
+ Assert.assertEquals(1, backfillCast.getIoConfig().getTaskCount());
+
+ // With overriding taskCount
+ final SupervisorSpec backfillSpecWithCount =
sourceSpec.createBackfillSpec("backfill-id-2", boundedStreamConfig, 5);
+ Assert.assertEquals("backfill-id-2", backfillSpecWithCount.getId());
+ final TestBackfillSupervisorSpec backfillWithCount =
(TestBackfillSupervisorSpec) backfillSpecWithCount;
+ Assert.assertEquals(5, backfillWithCount.getIoConfig().getTaskCount());
+ }
+
private static class TestSupervisorSpec implements SupervisorSpec
{
private final String id;
@@ -1137,4 +1328,103 @@ public class SupervisorManagerTest extends
EasyMockSupport
return Collections.singletonList(id);
}
}
+
+ @JsonTypeName("testBackfill")
+ private static class TestBackfillSupervisorSpec extends
SeekableStreamSupervisorSpec
+ {
+ @JsonCreator
+ TestBackfillSupervisorSpec(
+ @JsonProperty("id") String id,
+ @JsonProperty("spec") IngestionSpec ingestionSpec
+ )
+ {
+ super(
+ id,
+ ingestionSpec,
+ ImmutableMap.of("useConcurrentLocks", true),
+ false,
+ null, null, null, null,
+ MAPPER,
+ null, null, null, null
+ );
+ }
+
+ @Override
+ public Supervisor createSupervisor()
+ {
+ return null;
+ }
+
+ @Override
+ public String getType()
+ {
+ return "testBackfill";
+ }
+
+ @Override
+ public String getSource()
+ {
+ return "test-stream";
+ }
+
+ @Override
+ protected SeekableStreamSupervisorSpec toggleSuspend(boolean suspend)
+ {
+ return this;
+ }
+
+ @Override
+ public SeekableStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ return new TestBackfillSupervisorSpec(
+ backfillId,
+ new IngestionSpec(new IOConfig(getIoConfig().getStream(), taskCount,
boundedStreamConfig))
+ );
+ }
+
+ @Override
+ public SeekableStreamSupervisorIOConfig getIoConfig()
+ {
+ return getSpec().getIOConfig();
+ }
+
+ @JsonTypeName("testBackfillIngestionSpec")
+ static class IngestionSpec extends SeekableStreamSupervisorIngestionSpec
+ {
+ @JsonCreator
+ IngestionSpec(
+ @JsonProperty("ioConfig") IOConfig ioConfig
+ )
+ {
+ super(
+ new DataSchema(
+ "testDS",
+ new TimestampSpec("time", "auto", null),
+ new DimensionsSpec(Collections.emptyList()),
+ null, null, null, null, null
+ ),
+ ioConfig,
+ null
+ );
+ }
+ }
+
+ @JsonTypeName("testBackfillIOConfig")
+ static class IOConfig extends SeekableStreamSupervisorIOConfig
+ {
+ @JsonCreator
+ IOConfig(
+ @JsonProperty("stream") String stream,
+ @JsonProperty("taskCount") Integer taskCount,
+ @JsonProperty("boundedStreamConfig") BoundedStreamConfig
boundedStreamConfig
+ )
+ {
+ super(stream, null, 1, taskCount, null, null, null, false, null, null,
null, null, LagAggregator.DEFAULT, null, null, null, null, boundedStreamConfig);
+ }
+ }
+ }
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
index 4ccf4659994..31e0d604a22 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
@@ -34,6 +34,7 @@ import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAu
import
org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskClientFactory;
import
org.apache.druid.indexing.seekablestream.SeekableStreamStartSequenceNumbers;
import
org.apache.druid.indexing.seekablestream.TestSeekableStreamDataSourceMetadata;
+import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
@@ -1379,6 +1380,100 @@ public class SupervisorResourceTest extends
EasyMockSupport
verifyAll();
}
+ @Test
+ public void testResetToLatestAndBackfill()
+ {
+ // 200 - success
+
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+
EasyMock.expect(supervisorManager.getSupervisorIds()).andReturn(ImmutableSet.of("my-id"));
+ EasyMock.expect(supervisorManager.resetToLatestAndBackfill("my-id", null))
+ .andReturn(ImmutableMap.of("id", "my-id", "backfillSupervisorId",
"my-id_backfill_abcdefgh"));
+ replayAll();
+
+ Response response = supervisorResource.resetToLatestAndBackfill("my-id",
null);
+ Assert.assertEquals(200, response.getStatus());
+ Assert.assertEquals(
+ ImmutableMap.of("id", "my-id", "backfillSupervisorId",
"my-id_backfill_abcdefgh"),
+ response.getEntity()
+ );
+ verifyAll();
+ resetAll();
+
+ // 404 - supervisor does not exist
+
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+
EasyMock.expect(supervisorManager.getSupervisorIds()).andReturn(ImmutableSet.of());
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", null);
+ Assert.assertEquals(404, response.getStatus());
+ verifyAll();
+ resetAll();
+
+ // 400 - IAE (e.g. supervisor not running)
+
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+
EasyMock.expect(supervisorManager.getSupervisorIds()).andReturn(ImmutableSet.of("my-id"));
+ EasyMock.expect(supervisorManager.resetToLatestAndBackfill("my-id", null))
+ .andThrow(new IllegalArgumentException("Supervisor[my-id] must be
in a RUNNING state"));
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", null);
+ Assert.assertEquals(400, response.getStatus());
+ Assert.assertEquals(
+ ImmutableMap.of("error", "Supervisor[my-id] must be in a RUNNING
state"),
+ response.getEntity()
+ );
+ verifyAll();
+ resetAll();
+
+ // 500 - ISE (e.g. failed to retrieve offsets)
+
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+
EasyMock.expect(supervisorManager.getSupervisorIds()).andReturn(ImmutableSet.of("my-id"));
+ EasyMock.expect(supervisorManager.resetToLatestAndBackfill("my-id", null))
+ .andThrow(new IllegalStateException("Failed to get latest offsets
from stream"));
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", null);
+ Assert.assertEquals(500, response.getStatus());
+ Assert.assertEquals(
+ ImmutableMap.of("error", "Failed to get latest offsets from stream"),
+ response.getEntity()
+ );
+ verifyAll();
+ resetAll();
+
+ // 400 - invalid backfillTaskCount (zero)
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", 0);
+ Assert.assertEquals(400, response.getStatus());
+ Assert.assertEquals(
+ ImmutableMap.of("error", "backfillTaskCount must be a positive
integer"),
+ response.getEntity()
+ );
+ verifyAll();
+ resetAll();
+
+ // 400 - invalid backfillTaskCount (negative)
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", -1);
+ Assert.assertEquals(400, response.getStatus());
+ Assert.assertEquals(
+ ImmutableMap.of("error", "backfillTaskCount must be a positive
integer"),
+ response.getEntity()
+ );
+ verifyAll();
+ resetAll();
+
+ // 503 - no supervisor manager (not leader)
+
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent());
+ replayAll();
+
+ response = supervisorResource.resetToLatestAndBackfill("my-id", null);
+ Assert.assertEquals(503, response.getStatus());
+ verifyAll();
+ }
+
@Test
public void testNoopSupervisorSpecSerde() throws Exception
{
@@ -1668,6 +1763,16 @@ public class SupervisorResourceTest extends
EasyMockSupport
return null;
}
+ @Override
+ public SeekableStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ return null;
+ }
+
@JsonIgnore
@Nonnull
@Override
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
index 8d1b5350e8c..80120d07fdf 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
@@ -942,6 +942,16 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
return null;
}
+ @Override
+ public SeekableStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ return null;
+ }
+
@Override
public String getType()
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index d61049777c8..9e45920ad71 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -3059,7 +3059,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
final TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor()
{
@Override
- protected SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
+ public SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
String stream,
Map<String, String> map
)
@@ -3284,7 +3284,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
}
@Override
- protected void updatePartitionLagFromStream()
+ public void updatePartitionLagFromStream()
{
// do nothing
}
@@ -3381,7 +3381,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
}
@Override
- protected SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
+ public SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
String stream,
Map<String, String> map
)
@@ -3521,7 +3521,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
}
@Override
- protected Map<String, String> getLatestSequencesFromStream()
+ public Map<String, String> getLatestSequencesFromStream()
{
return streamOffsets;
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
index 4eefaed9bd9..c96a64211b9 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
@@ -124,7 +124,7 @@ public abstract class SeekableStreamSupervisorTestBase
}
@Override
- protected void updatePartitionLagFromStream()
+ public void updatePartitionLagFromStream()
{
// do nothing
}
@@ -205,7 +205,7 @@ public abstract class SeekableStreamSupervisorTestBase
}
@Override
- protected SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
+ public SeekableStreamDataSourceMetadata<String, String>
createDataSourceMetaDataForReset(
String stream,
Map<String, String> map
)
@@ -436,6 +436,16 @@ public abstract class SeekableStreamSupervisorTestBase
{
return null;
}
+
+ @Override
+ public SeekableStreamSupervisorSpec createBackfillSpec(
+ String backfillId,
+ BoundedStreamConfig boundedStreamConfig,
+ @Nullable Integer taskCount
+ )
+ {
+ return null;
+ }
}
protected static SeekableStreamSupervisorTuningConfig getTuningConfig()
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
index 0ca643f109f..35488be081b 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
@@ -322,6 +322,12 @@ public class MSQWorkerTaskLauncherRetryTest
throw new UOE("Not implemented");
}
+ @Override
+ public ListenableFuture<Map<String, Object>>
resetToLatestAndBackfill(String supervisorId)
+ {
+ throw new UOE("Not implemented");
+ }
+
@Override
public ListenableFuture<CloseableIterator<SupervisorStatus>>
supervisorStatuses()
{
diff --git
a/server/src/main/java/org/apache/druid/rpc/indexing/NoopOverlordClient.java
b/server/src/main/java/org/apache/druid/rpc/indexing/NoopOverlordClient.java
index 81fccf19f13..2b1ad6a555a 100644
--- a/server/src/main/java/org/apache/druid/rpc/indexing/NoopOverlordClient.java
+++ b/server/src/main/java/org/apache/druid/rpc/indexing/NoopOverlordClient.java
@@ -114,6 +114,12 @@ public class NoopOverlordClient implements OverlordClient
throw new UnsupportedOperationException();
}
+ @Override
+ public ListenableFuture<Map<String, Object>> resetToLatestAndBackfill(String
supervisorId)
+ {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public ListenableFuture<CloseableIterator<SupervisorStatus>>
supervisorStatuses()
{
diff --git
a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
index c4d34899777..baf7e4297c9 100644
--- a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
+++ b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
@@ -197,6 +197,15 @@ public interface OverlordClient
*/
ListenableFuture<Map<String, String>> terminateSupervisor(String
supervisorId);
+ /**
+ * Resets a supervisor to the latest stream offsets and starts a bounded
backfill supervisor.
+ * <p>
+ * API: {@code POST
/druid/indexer/v1/supervisor/<id>/resetToLatestAndBackfill}
+ *
+ * @return Map containing "id" and "backfillSupervisorId"
+ */
+ ListenableFuture<Map<String, Object>> resetToLatestAndBackfill(String
supervisorId);
+
/**
* Returns all current supervisor statuses.
*/
diff --git
a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClientImpl.java
b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClientImpl.java
index 0499a62f090..3657d8b83a6 100644
--- a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClientImpl.java
+++ b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClientImpl.java
@@ -265,6 +265,23 @@ public class OverlordClientImpl implements OverlordClient
);
}
+ @Override
+ public ListenableFuture<Map<String, Object>> resetToLatestAndBackfill(String
supervisorId)
+ {
+ final String path = StringUtils.format(
+ "/druid/indexer/v1/supervisor/%s/resetToLatestAndBackfill",
+ StringUtils.urlEncode(supervisorId)
+ );
+
+ return FutureUtils.transform(
+ client.asyncRequest(
+ new RequestBuilder(HttpMethod.POST, path),
+ new BytesFullResponseHandler()
+ ),
+ holder -> JacksonUtils.readValue(jsonMapper, holder.getContent(), new
TypeReference<>() {})
+ );
+ }
+
@Override
public ListenableFuture<CloseableIterator<SupervisorStatus>>
supervisorStatuses()
{
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
index 257533aecbd..6ae8750b8d8 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
@@ -430,6 +430,16 @@ public class EmbeddedClusterApis implements
EmbeddedResource
return onLeaderOverlord(o -> o.postSupervisor(supervisor)).get("id");
}
+ /**
+ * Resets a supervisor to the latest stream offsets and starts a bounded
backfill supervisor.
+ *
+ * @return Map containing "id" and "backfillSupervisorId"
+ */
+ public Map<String, Object> resetToLatestAndBackfill(String supervisorId)
+ {
+ return onLeaderOverlord(o -> o.resetToLatestAndBackfill(supervisorId));
+ }
+
/**
* Fetches the current status of the given supervisor ID.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
