You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "abhishekrb19 (via GitHub)" <gi...@apache.org> on 2023/08/07 22:35:59 UTC
[PR] Reset offsets supervisor API (druid)
abhishekrb19 opened a new pull request, #14772:
URL: https://github.com/apache/druid/pull/14772
This PR adds a new API: `/druid/indexer/v1/supervisor/<supervisorId>/resetOffsets`. This API accepts a reset datasource metadata in the body and only resets the specified partition offsets in the metadata store. Unlike the `/reset` API, this new API retains any previously checkpointed partition offsets not specified in the payload, so tasks will resume from where they left off. Further, only tasks serving the reset partitions will be restarted.
<!-- 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
A few use cases:
- A supervisor that is reading from an offset that has fallen off (moved past the retention period or similar), so instead of a `/reset` that only does an earliest or latest offset reset, users will now be able to specify specific partition offsets.
- Users wanting some history for a streaming datasource but not all the way to earliest can also benefit from this API
<!-- 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
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
Users can reset specific partition offsets for a supervisor using the `resetOffsets` API.
<hr>
##### Key changed/added classes in this PR
* `SupervisorResource.java`
* `Supervisor.java`
* `SeekableStreamSupervisor.java`
* `SeekableStreamSupervisorStateTest.java`
* `TestSeekableStreamDataSourceMetadata.java`
<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.
- [ ] 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.)
- [x] 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.
- [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.
- [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] Reset offsets supervisor API (druid)
Posted by "vogievetsky (via GitHub)" <gi...@apache.org>.
vogievetsky merged PR #14772:
URL: https://github.com/apache/druid/pull/14772
--
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] Reset offsets supervisor API (druid)
Posted by "vogievetsky (via GitHub)" <gi...@apache.org>.
vogievetsky commented on PR #14772:
URL: https://github.com/apache/druid/pull/14772#issuecomment-1682989251
Ignoring the CodeQL issues per the comment above
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1291757329
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
+ throw InvalidInput.exception(
+ "Stream[%s] doesn't exist in the supervisor[%s]. Supervisor is consuming stream[%s].",
+ resetStream,
+ supervisorId,
+ ioConfig.getStream()
+ );
+ }
+ for (Entry<PartitionIdType, SequenceOffsetType> resetPartitionOffset : resetMetadata
+ .getSeekableStreamSequenceNumbers()
+ .getPartitionSequenceNumberMap()
+ .entrySet()) {
+ final TaskGroup taskGroup = activelyReadingTaskGroups.get(getTaskGroupIdForPartition(resetPartitionOffset.getKey()));
+ if (taskGroup == null) {
+ throw InvalidInput.exception("No task found serving stream[%s], partition[%s] in supervisor[%s].",
+ resetStream, resetPartitionOffset.getKey(), supervisorId);
+ } else if (!taskGroup.startingSequences.containsKey(resetPartitionOffset.getKey())) {
+ throw InvalidInput.exception(
+ "Partition[%s] doesn't exist in checkpointed metadata for stream[%s] and supervisor[%s]."
Review Comment:
Good point, removed this check
--
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] Reset offsets supervisor API (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1286637294
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -277,6 +277,12 @@ public void reset(DataSourceMetadata dataSourceMetadata)
}
}
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata, boolean setOffsetsInMetadata)
Review Comment:
This method seems to be called with`setOffsetsInMetadata` always set to true
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on PR #14772:
URL: https://github.com/apache/druid/pull/14772#issuecomment-1676653036
The CodeQL alert about log entries depending on a user-provided value appears to be flagging existing logging code that's not part of this PR. I think it's a false positive alert.
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1286676667
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -277,6 +277,12 @@ public void reset(DataSourceMetadata dataSourceMetadata)
}
}
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata, boolean setOffsetsInMetadata)
Review Comment:
yeah, thanks, cleaned up. This was a leftover from some refactoring. `ResetOffsets` always sets it to true and only `ResetNotice()` and `resetInternal()` have the additional parameter.
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1286468107
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaDataSourceMetadataTest.java:
##########
@@ -170,4 +211,23 @@
{
return new KafkaDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>("foo", offsets));
}
+
+ private static ObjectMapper createObjectMapper()
+ {
+ DruidModule module = new KafkaIndexTaskModule();
+ Injector injector = Initialization.makeInjectorWithModules(
+ GuiceInjectors.makeStartupInjector(),
+ Arrays.asList(
+ module,
+ (Module) binder -> {
+ binder.bindConstant().annotatedWith(Names.named("serviceName")).to("test");
+ binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8000);
+ binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9000);
+ }
+ )
+ );
Review Comment:
Fixed
--
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] Reset offsets supervisor API (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1286455775
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -1392,7 +1736,7 @@
@Override
protected int getTaskGroupIdForPartition(String partition)
{
- return 0;
+ return Integer.parseInt(partition) % spec.getIoConfig().getTaskCount();
Review Comment:
## Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/5418)
##########
extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisDataSourceMetadataTest.java:
##########
@@ -233,4 +274,23 @@
{
return new KinesisDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>("foo", sequences));
}
+
+ private static ObjectMapper createObjectMapper()
+ {
+ DruidModule module = new KinesisIndexingServiceModule();
+ Injector injector = Initialization.makeInjectorWithModules(
+ GuiceInjectors.makeStartupInjector(),
+ Arrays.asList(
+ module,
+ (Module) binder -> {
+ binder.bindConstant().annotatedWith(Names.named("serviceName")).to("test");
+ binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8000);
+ binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9000);
+ }
+ )
+ );
Review Comment:
## Deprecated method or constructor invocation
Invoking [Initialization.makeInjectorWithModules](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/5417)
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaDataSourceMetadataTest.java:
##########
@@ -170,4 +211,23 @@
{
return new KafkaDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>("foo", offsets));
}
+
+ private static ObjectMapper createObjectMapper()
+ {
+ DruidModule module = new KafkaIndexTaskModule();
+ Injector injector = Initialization.makeInjectorWithModules(
+ GuiceInjectors.makeStartupInjector(),
+ Arrays.asList(
+ module,
+ (Module) binder -> {
+ binder.bindConstant().annotatedWith(Names.named("serviceName")).to("test");
+ binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8000);
+ binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9000);
+ }
+ )
+ );
Review Comment:
## Deprecated method or constructor invocation
Invoking [Initialization.makeInjectorWithModules](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/5416)
--
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] Reset offsets supervisor API (druid)
Posted by "vogievetsky (via GitHub)" <gi...@apache.org>.
vogievetsky commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1295119337
##########
docs/api-reference/supervisor-api.md:
##########
@@ -3030,6 +3030,86 @@ Host: http://ROUTER_IP:ROUTER_PORT
```
</details>
+### Reset Offsets for a supervisor
+
+Resets the specified offsets for a supervisor. This endpoint clears _only_ the specified offsets in Kafka or sequence numbers in Kinesis, prompting the supervisor to resume data reading.
+The supervisor will start from the specified reset offsets for the partitions specified and for the other partitions from the stored offset. It kills and recreates active tasks pertaining to
+the partitions specified to read from valid offsets.
+
+Use this endpoint to selectively reset offsets for partitions without resetting the entire set.
+
+#### URL
+
+<code class="postAPI">POST</code> <code>/druid/indexer/v1/supervisor/:supervisorId/resetOffsets</code>
+
+#### Responses
+
+<!--DOCUSAURUS_CODE_TABS-->
+
+<!--200 SUCCESS-->
+
+*Successfully reset offsets*
+
+<!--404 NOT FOUND-->
+
+*Invalid supervisor ID*
+
+<!--END_DOCUSAURUS_CODE_TABS-->
+
+---
+
+#### Sample request
+
+The following example shows how to reset offsets for a kafka supervisor with the name `social_media`. Let's say the supervisor is reading
+from a kafka topic `ads_media_stream` and has the stored offsets: `{"0": 0, "1": 10, "2": 20, "3": 40}`.
+
+<!--DOCUSAURUS_CODE_TABS-->
+
+<!--cURL-->
+
+```shell
+curl --request POST "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/supervisor/social_media/resetOffsets"
+--header 'Content-Type: application/json'
+--data-raw '{"type":"kafka","partitions":{"type":"end","stream":"ads_media_stream","partitionOffsetMap":{"0":100, "2": 650}}}'
+```
+
+The above operation will reset offsets only for partitions 0 and 2 to 100 and 650 respectively. After a successful reset,
+when the supervisor's tasks restart, they will resume reading from `{"0": 100, "1": 10, "2": 650, "3": 40}`.
+
+<!--HTTP-->
+
+```HTTP
+POST /druid/indexer/v1/supervisor/social_media/resetOffsets HTTP/1.1
+Host: http://ROUTER_IP:ROUTER_PORT
+Content-Type: application/json
+
+{
+ "type": "kafka",
+ "partitions": {
+ "type": "end",
Review Comment:
What is `"type": "end",` ? where is it documented?
--
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] Reset offsets supervisor API (druid)
Posted by "jon-wei (via GitHub)" <gi...@apache.org>.
jon-wei commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1287787896
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
Review Comment:
nit: resetStream is already declared above and can be used in this if
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -3614,7 +3692,8 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
if (!checkOffsetAvailability(partition, sequence)) {
if (taskTuningConfig.isResetOffsetAutomatically()) {
resetInternal(
- createDataSourceMetaDataForReset(ioConfig.getStream(), ImmutableMap.of(partition, sequence))
+ createDataSourceMetaDataForReset(ioConfig.getStream(), ImmutableMap.of(partition, sequence)),
+ false
Review Comment:
what does the `false` parameter here mean?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
+ throw InvalidInput.exception(
+ "Stream[%s] doesn't exist in the supervisor[%s]. Supervisor is consuming stream[%s].",
+ resetStream,
+ supervisorId,
+ ioConfig.getStream()
+ );
+ }
+ for (Entry<PartitionIdType, SequenceOffsetType> resetPartitionOffset : resetMetadata
+ .getSeekableStreamSequenceNumbers()
+ .getPartitionSequenceNumberMap()
+ .entrySet()) {
+ final TaskGroup taskGroup = activelyReadingTaskGroups.get(getTaskGroupIdForPartition(resetPartitionOffset.getKey()));
Review Comment:
if the supervisor is suspended, would there be any activelyReadingTaskGroups, and would that block the reset?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
+ throw InvalidInput.exception(
+ "Stream[%s] doesn't exist in the supervisor[%s]. Supervisor is consuming stream[%s].",
+ resetStream,
+ supervisorId,
+ ioConfig.getStream()
+ );
+ }
+ for (Entry<PartitionIdType, SequenceOffsetType> resetPartitionOffset : resetMetadata
+ .getSeekableStreamSequenceNumbers()
+ .getPartitionSequenceNumberMap()
+ .entrySet()) {
+ final TaskGroup taskGroup = activelyReadingTaskGroups.get(getTaskGroupIdForPartition(resetPartitionOffset.getKey()));
+ if (taskGroup == null) {
+ throw InvalidInput.exception("No task found serving stream[%s], partition[%s] in supervisor[%s].",
+ resetStream, resetPartitionOffset.getKey(), supervisorId);
+ } else if (!taskGroup.startingSequences.containsKey(resetPartitionOffset.getKey())) {
+ throw InvalidInput.exception(
+ "Partition[%s] doesn't exist in checkpointed metadata for stream[%s] and supervisor[%s]."
Review Comment:
Do we want to enforce that the partition must exist in the current checkpoint? (e.g., maybe someone repartitioned their stream and then wants to start from a specific offset in the newly added partitions)
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1291921303
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
+ throw InvalidInput.exception(
+ "Stream[%s] doesn't exist in the supervisor[%s]. Supervisor is consuming stream[%s].",
+ resetStream,
+ supervisorId,
+ ioConfig.getStream()
+ );
+ }
+ for (Entry<PartitionIdType, SequenceOffsetType> resetPartitionOffset : resetMetadata
+ .getSeekableStreamSequenceNumbers()
+ .getPartitionSequenceNumberMap()
+ .entrySet()) {
+ final TaskGroup taskGroup = activelyReadingTaskGroups.get(getTaskGroupIdForPartition(resetPartitionOffset.getKey()));
Review Comment:
Adjusted based on our discussions, thanks! A user can issue a reset offsets even when the supervisor is suspended
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1286676667
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -277,6 +277,12 @@ public void reset(DataSourceMetadata dataSourceMetadata)
}
}
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata, boolean setOffsetsInMetadata)
Review Comment:
yeah, thanks, cleaned up. This was a leftover from some refactoring. `resetOffsets` always sets it to true and only `ResetNotice()` and `resetInternal()` have the additional parameter.
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1291755312
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -998,10 +1006,66 @@ public void stop(boolean stopGracefully)
}
@Override
- public void reset(DataSourceMetadata dataSourceMetadata)
+ public void reset(@Nullable final DataSourceMetadata dataSourceMetadata)
+ {
+ log.info("Posting ResetNotice with datasource metadata [%s]", dataSourceMetadata);
+ addNotice(new ResetNotice(dataSourceMetadata, false));
+ }
+
+ /**
+ * Reset offsets with provided dataSource metadata. Validates {@code resetDataSourceMetadata},
+ * creates a {@code ResetNotice} with the metadata and adds it to the notice queue. The resulting stored offsets
+ * is a union of existing checkpointed offsets with provided offsets.
+ * @param resetDataSourceMetadata required datasource metadata with offsets to reset.
+ * @throws DruidException if any metadata attribute doesn't match the supervisor's.
+ */
+ @Override
+ public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
{
- log.info("Posting ResetNotice");
- addNotice(new ResetNotice(dataSourceMetadata));
+ if (resetDataSourceMetadata == null) {
+ throw InvalidInput.exception("Reset dataSourceMetadata is required for resetOffsets.");
+ }
+
+ if (!checkSourceMetadataMatch(resetDataSourceMetadata)) {
+ throw InvalidInput.exception(
+ "Datasource metadata instance does not match required, found instance of [%s].",
+ resetDataSourceMetadata.getClass()
+ );
+ }
+ @SuppressWarnings("unchecked")
+ final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) resetDataSourceMetadata;
+ String resetStream = resetMetadata.getSeekableStreamSequenceNumbers().getStream();
+ if (!resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {
Review Comment:
yep, done
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1296372703
##########
docs/api-reference/supervisor-api.md:
##########
@@ -3030,6 +3030,86 @@ Host: http://ROUTER_IP:ROUTER_PORT
```
</details>
+### Reset Offsets for a supervisor
+
+Resets the specified offsets for a supervisor. This endpoint clears _only_ the specified offsets in Kafka or sequence numbers in Kinesis, prompting the supervisor to resume data reading.
+The supervisor will start from the specified reset offsets for the partitions specified and for the other partitions from the stored offset. It kills and recreates active tasks pertaining to
+the partitions specified to read from valid offsets.
+
+Use this endpoint to selectively reset offsets for partitions without resetting the entire set.
+
+#### URL
+
+<code class="postAPI">POST</code> <code>/druid/indexer/v1/supervisor/:supervisorId/resetOffsets</code>
+
+#### Responses
+
+<!--DOCUSAURUS_CODE_TABS-->
+
+<!--200 SUCCESS-->
+
+*Successfully reset offsets*
+
+<!--404 NOT FOUND-->
+
+*Invalid supervisor ID*
+
+<!--END_DOCUSAURUS_CODE_TABS-->
+
+---
+
+#### Sample request
+
+The following example shows how to reset offsets for a kafka supervisor with the name `social_media`. Let's say the supervisor is reading
+from a kafka topic `ads_media_stream` and has the stored offsets: `{"0": 0, "1": 10, "2": 20, "3": 40}`.
+
+<!--DOCUSAURUS_CODE_TABS-->
+
+<!--cURL-->
+
+```shell
+curl --request POST "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/supervisor/social_media/resetOffsets"
+--header 'Content-Type: application/json'
+--data-raw '{"type":"kafka","partitions":{"type":"end","stream":"ads_media_stream","partitionOffsetMap":{"0":100, "2": 650}}}'
+```
+
+The above operation will reset offsets only for partitions 0 and 2 to 100 and 650 respectively. After a successful reset,
+when the supervisor's tasks restart, they will resume reading from `{"0": 100, "1": 10, "2": 650, "3": 40}`.
+
+<!--HTTP-->
+
+```HTTP
+POST /druid/indexer/v1/supervisor/social_media/resetOffsets HTTP/1.1
+Host: http://ROUTER_IP:ROUTER_PORT
+Content-Type: application/json
+
+{
+ "type": "kafka",
+ "partitions": {
+ "type": "end",
Review Comment:
Added docs for the payload. Hope that clarifies it
--
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] Reset offsets supervisor API (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #14772:
URL: https://github.com/apache/druid/pull/14772#discussion_r1291758237
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -3614,7 +3692,8 @@ private OrderedSequenceNumber<SequenceOffsetType> getOffsetFromStorageForPartiti
if (!checkOffsetAvailability(partition, sequence)) {
if (taskTuningConfig.isResetOffsetAutomatically()) {
resetInternal(
- createDataSourceMetaDataForReset(ioConfig.getStream(), ImmutableMap.of(partition, sequence))
+ createDataSourceMetaDataForReset(ioConfig.getStream(), ImmutableMap.of(partition, sequence)),
+ false
Review Comment:
I've pulled out the logic to its separate method
--
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