You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "adithyachakilam (via GitHub)" <gi...@apache.org> on 2024/11/03 08:42:28 UTC
[PR] supervisor: propagate rejection periods to tasks (druid)
adithyachakilam opened a new pull request, #17442:
URL: https://github.com/apache/druid/pull/17442
### Description
In cases where `stopTasksCount` is configured which makes streaming tasks run much longer (in some cases) than the configured task duration, early/late rejection periods are not correctly implemented since they are configured during the task startup time in IOConfig. Instead, this PR propagates them in to the task so that they could be computed in real-time.
Also, to avoid computation on each row, a background thread is scheduled to updated those min/max times once every 10 minutes.
<hr>
##### Key changed/added classes in this PR
* `SeekableStreamIndexTaskRunner`
* `SeekableStreamIndexTaskIOConfig`
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.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [x] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] 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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1829910832
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskIOConfig.java:
##########
@@ -51,7 +53,8 @@ public SeekableStreamIndexTaskIOConfig(
final Boolean useTransaction,
final DateTime minimumMessageTime,
final DateTime maximumMessageTime,
- @Nullable final InputFormat inputFormat
+ @Nullable final InputFormat inputFormat,
+ @Nullable final Duration taskDuration // can be null for backward compabitility
Review Comment:
I think its good to have Duration and no need to convert back and forth and also confusion of adding `mins` to the variable. You think there are any issues with the schema deserializing ? (I see at least `Period` used, maybe we can switch to it if so )
Also, regarding variable name change, Do you think there would be a usecase of user editing it to a different value ? I would prefer to use `taskDuration` so that no one touches and would also avoid the need for docs (Also, I dont find this documented anywhere, can you point me to the file ?)
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1829918957
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -2092,4 +2109,47 @@ protected abstract void possiblyResetDataSourceMetadata(
protected abstract boolean isEndOffsetExclusive();
protected abstract TypeReference<List<SequenceMetadata<PartitionIdType, SequenceOffsetType>>> getSequenceMetadataTypeReference();
+
+
+ private void addTaskDurationToMinMaxTimes()
+ {
+ if (minMessageTime.isPresent()) {
+ minMessageTime = Optional.of(minMessageTime.get()
Review Comment:
Yeah its ok.
Its complaining because we are using the same variable to update the value but we only do it once every taskDuration. If it was done more frequently we should have used atomic references.
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1847610129
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
Who closes this exec service ?
--
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] supervisor: propagate rejection periods to tasks (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1827012102
##########
extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java:
##########
@@ -50,8 +50,8 @@ public RabbitStreamIndexTaskIOConfig(
@JsonProperty("consumerProperties") Map<String, Object> consumerProperties,
@JsonProperty("pollTimeout") Long pollTimeout,
@JsonProperty("useTransaction") Boolean useTransaction,
- @JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
- @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
+ @JsonProperty("lateMessageRejectionPeriod") Duration lateMessageRejectionPeriod,
+ @JsonProperty("earlyMessageRejectionPeriod") Duration earlyMessageRejectionPeriod,
Review Comment:
May to avoid all these issues, How about instead of propagating periods, Is it okay if we just recalculate the period min/max time (and avoid introducing these new fields)?
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on PR #17442:
URL: https://github.com/apache/druid/pull/17442#issuecomment-2484140355
Overruling CI. I had to re-trigger a bunch of failed ITs, and now they have all passed. The only remaining failures are code coverage.
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1847619827
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
cc @adithyachakilam
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1829740240
##########
extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTest.java:
##########
@@ -366,4 +368,47 @@ public void testReportPayload()
Assert.assertEquals(30 * 60, payload.getDurationSeconds());
}
+ @Test
+ public void testCreateTaskIOConfig()
+ {
+ supervisor = getSupervisor(
+ 1,
+ 1,
+ false,
+ "PT30M",
+ null,
+ null,
+ RabbitStreamSupervisorTest.dataSchema,
+ tuningConfig
+ );
+
+ Assert.assertEquals(supervisor.createTaskIoConfig(
Review Comment:
Can you split this into 2 lines - one to get the actual task duration, and another to do the assert please.
The assert order should be reversed ie assertEquals(<expencted>, <actual>)
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskIOConfig.java:
##########
@@ -134,4 +138,9 @@ public InputFormat getInputFormat()
{
return inputFormat;
}
+
+ public Duration getTaskDuration()
Review Comment:
Looks like this should be marked as nullable
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskIOConfig.java:
##########
@@ -51,7 +53,8 @@ public SeekableStreamIndexTaskIOConfig(
final Boolean useTransaction,
final DateTime minimumMessageTime,
final DateTime maximumMessageTime,
- @Nullable final InputFormat inputFormat
+ @Nullable final InputFormat inputFormat,
+ @Nullable final Duration taskDuration // can be null for backward compabitility
Review Comment:
Is there an advantage to using Duration? Would it be simpler to pass this in as a long of minutes
```suggestion
@Nullable final Long refreshRejectionPeriodsInMinutes // can be null for backward compabitility
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -2092,4 +2109,47 @@ protected abstract void possiblyResetDataSourceMetadata(
protected abstract boolean isEndOffsetExclusive();
protected abstract TypeReference<List<SequenceMetadata<PartitionIdType, SequenceOffsetType>>> getSequenceMetadataTypeReference();
+
+
+ private void addTaskDurationToMinMaxTimes()
+ {
+ if (minMessageTime.isPresent()) {
+ minMessageTime = Optional.of(minMessageTime.get()
+ .plusMinutes(ioConfig.getTaskDuration()
+ .toStandardMinutes()
+ .getMinutes()));
+ }
+
+ if (maxMessageTime.isPresent()) {
+ maxMessageTime = Optional.of(maxMessageTime.get()
+ .plusMinutes(ioConfig.getTaskDuration()
+ .toStandardMinutes()
+ .getMinutes()));
+ }
+ }
+
+ public boolean withinMinMaxRecordTime(final InputRow row)
+ {
+ final boolean beforeMinimumMessageTime = minMessageTime.isPresent() &&
+ minMessageTime.get().isAfter(row.getTimestamp());
+ final boolean afterMaximumMessageTime = maxMessageTime.isPresent() &&
+ maxMessageTime.get().isBefore(row.getTimestamp());
+
+ if (log.isDebugEnabled()) {
+ if (beforeMinimumMessageTime) {
+ log.debug(
+ "CurrentTimeStamp[%s] is before MinimumMessageTime[%s]",
+ row.getTimestamp(),
+ ioConfig.getMinimumMessageTime().get()
Review Comment:
```suggestion
minMessageTime
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -2092,4 +2109,47 @@ protected abstract void possiblyResetDataSourceMetadata(
protected abstract boolean isEndOffsetExclusive();
protected abstract TypeReference<List<SequenceMetadata<PartitionIdType, SequenceOffsetType>>> getSequenceMetadataTypeReference();
+
+
+ private void addTaskDurationToMinMaxTimes()
+ {
+ if (minMessageTime.isPresent()) {
+ minMessageTime = Optional.of(minMessageTime.get()
+ .plusMinutes(ioConfig.getTaskDuration()
+ .toStandardMinutes()
+ .getMinutes()));
+ }
+
+ if (maxMessageTime.isPresent()) {
+ maxMessageTime = Optional.of(maxMessageTime.get()
+ .plusMinutes(ioConfig.getTaskDuration()
+ .toStandardMinutes()
+ .getMinutes()));
+ }
+ }
+
+ public boolean withinMinMaxRecordTime(final InputRow row)
+ {
+ final boolean beforeMinimumMessageTime = minMessageTime.isPresent() &&
+ minMessageTime.get().isAfter(row.getTimestamp());
+ final boolean afterMaximumMessageTime = maxMessageTime.isPresent() &&
+ maxMessageTime.get().isBefore(row.getTimestamp());
+
+ if (log.isDebugEnabled()) {
+ if (beforeMinimumMessageTime) {
+ log.debug(
+ "CurrentTimeStamp[%s] is before MinimumMessageTime[%s]",
+ row.getTimestamp(),
+ ioConfig.getMinimumMessageTime().get()
+ );
+ } else if (afterMaximumMessageTime) {
+ log.debug(
+ "CurrentTimeStamp[%s] is after MaximumMessageTime[%s]",
+ row.getTimestamp(),
+ ioConfig.getMaximumMessageTime().get()
Review Comment:
```suggestion
maxMessageTime
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -2092,4 +2109,47 @@ protected abstract void possiblyResetDataSourceMetadata(
protected abstract boolean isEndOffsetExclusive();
protected abstract TypeReference<List<SequenceMetadata<PartitionIdType, SequenceOffsetType>>> getSequenceMetadataTypeReference();
+
+
+ private void addTaskDurationToMinMaxTimes()
+ {
+ if (minMessageTime.isPresent()) {
+ minMessageTime = Optional.of(minMessageTime.get()
Review Comment:
Is it ok that intelliJ is warning that this is an unsafe update of a volatile variable?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java:
##########
@@ -43,16 +43,20 @@ public class KafkaConsumerMonitor extends AbstractMonitor
ImmutableMap.<String, String>builder()
.put("bytes-consumed-total", "kafka/consumer/bytesConsumed")
.put("records-consumed-total", "kafka/consumer/recordsConsumed")
+ .put("io-wait-time-ns-total", "kafka/consumer/io/time")
Review Comment:
unintended change from another patch
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -246,6 +248,9 @@ public enum Status
private final Map<PartitionIdType, Long> partitionsThroughput = new HashMap<>();
+ private volatile Optional<DateTime> minMessageTime;
+ private volatile Optional<DateTime> maxMessageTime;
Review Comment:
nit: Instead of Optional, these could be DateTime objects and they could be instantiated like which would simplify the code slightly
```
minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
```
--
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] supervisor: propagate rejection periods to tasks (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on PR #17442:
URL: https://github.com/apache/druid/pull/17442#issuecomment-2453504179
@suneet-s Thanks for the review, I've changed the patch to just update the min/max times once every task duration. So this also avoids all those version problems.
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s merged PR #17442:
URL: https://github.com/apache/druid/pull/17442
--
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] supervisor: propagate rejection periods to tasks (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1827001112
##########
extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java:
##########
@@ -50,8 +50,8 @@ public RabbitStreamIndexTaskIOConfig(
@JsonProperty("consumerProperties") Map<String, Object> consumerProperties,
@JsonProperty("pollTimeout") Long pollTimeout,
@JsonProperty("useTransaction") Boolean useTransaction,
- @JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
- @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
+ @JsonProperty("lateMessageRejectionPeriod") Duration lateMessageRejectionPeriod,
+ @JsonProperty("earlyMessageRejectionPeriod") Duration earlyMessageRejectionPeriod,
Review Comment:
These new fields need to be in addition to the old fields to ensure a newer version of the overlord works with older versions of the indexing tasks. But more importantly, in the case of a rollback - an older overlord needs to work with a newer version of indexing services
--
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] supervisor: propagate rejection periods to tasks (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1827012102
##########
extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java:
##########
@@ -50,8 +50,8 @@ public RabbitStreamIndexTaskIOConfig(
@JsonProperty("consumerProperties") Map<String, Object> consumerProperties,
@JsonProperty("pollTimeout") Long pollTimeout,
@JsonProperty("useTransaction") Boolean useTransaction,
- @JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
- @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
+ @JsonProperty("lateMessageRejectionPeriod") Duration lateMessageRejectionPeriod,
+ @JsonProperty("earlyMessageRejectionPeriod") Duration earlyMessageRejectionPeriod,
Review Comment:
May to avoid all these issues, How about instead of propagating periods, Is it okay if we just recalculate the period min/max time (and avoid introducing these new fields)?
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1849752324
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
> This seems to be creating on thread for each supervisor. Seems a bit much no ? Can we use a common exec service for this ?
+1
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1841611502
##########
indexing-service/src/test/resources/log4j2.xml:
##########
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+
+<Configuration status="WARN">
+ <Loggers>
+ <Logger level="debug" name="org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner" additivity="false">
+ <AppenderRef ref="Console"/>
+ </Logger>
+ </Loggers>
+</Configuration>
Review Comment:
Is this change needed?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -2092,4 +2108,33 @@ protected abstract void possiblyResetDataSourceMetadata(
protected abstract boolean isEndOffsetExclusive();
protected abstract TypeReference<List<SequenceMetadata<PartitionIdType, SequenceOffsetType>>> getSequenceMetadataTypeReference();
+
+ private void refreshMinMaxMessageTime()
+ {
+ minMessageTime = minMessageTime.plusMinutes(ioConfig.getRefreshRejectionPeriodsInMinutes().intValue());
Review Comment:
Let's add an info log here. This seems like it would be important to know when the min max times are refreshed
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1841668533
##########
indexing-service/src/test/resources/log4j2.xml:
##########
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+
+<Configuration status="WARN">
+ <Loggers>
+ <Logger level="debug" name="org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner" additivity="false">
+ <AppenderRef ref="Console"/>
+ </Logger>
+ </Loggers>
+</Configuration>
Review Comment:
It atleast works locally for coverage.
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1847619477
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
This seems to be creating on thread for each supervisor. Seems a bit much no ? Can we use a common exec service for this ?
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1849847687
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
this is creating thread on task not on supervisor ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] supervisor: propagate rejection periods to tasks (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1827034171
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.InputRowParser;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexing.common.LockGranularity;
+import org.apache.druid.indexing.common.TaskToolbox;
+import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord;
+import org.apache.druid.indexing.seekablestream.common.OrderedSequenceNumber;
+import org.apache.druid.indexing.seekablestream.common.RecordSupplier;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SeekableStreamIndexTaskRunnerTest
+{
+ @Mock
+ private InputRow row;
+
+ @Mock
+ private SeekableStreamIndexTask task;
+
+ @Test
+ public void testWithinMinMaxTime()
+ {
+ DimensionsSpec dimensionsSpec = new DimensionsSpec(
+ Arrays.asList(
+ new StringDimensionSchema("d1"),
+ new StringDimensionSchema("d2")
+ )
+ );
+ DataSchema schema =
+ DataSchema.builder()
+ .withDataSource("datasource")
+ .withTimestamp(new TimestampSpec(null, null, null))
+ .withDimensions(dimensionsSpec)
+ .withGranularity(
+ new UniformGranularitySpec(Granularities.MINUTE, Granularities.NONE, null)
+ )
+ .build();
+
+ SeekableStreamIndexTaskTuningConfig tuningConfig = Mockito.mock(SeekableStreamIndexTaskTuningConfig.class);
+ SeekableStreamIndexTaskIOConfig<String, String> ioConfig = Mockito.mock(SeekableStreamIndexTaskIOConfig.class);
+ SeekableStreamStartSequenceNumbers<String, String> sequenceNumbers = Mockito.mock(SeekableStreamStartSequenceNumbers.class);
+ SeekableStreamEndSequenceNumbers<String, String> endSequenceNumbers = Mockito.mock(SeekableStreamEndSequenceNumbers.class);
+
+ DateTime now = DateTimes.nowUtc();
+
+ Mockito.when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(2));
+ Mockito.when(ioConfig.getMaximumMessageTime()).thenReturn(Optional.of(DateTimes.nowUtc().plusHours(2)));
+ Mockito.when(ioConfig.getMinimumMessageTime()).thenReturn(Optional.of(DateTimes.nowUtc().minusHours(2)));
+ Mockito.when(ioConfig.getInputFormat()).thenReturn(new JsonInputFormat(null, null, null, null, null));
+ Mockito.when(ioConfig.getStartSequenceNumbers()).thenReturn(sequenceNumbers);
+ Mockito.when(ioConfig.getEndSequenceNumbers()).thenReturn(endSequenceNumbers);
+
+ Mockito.when(endSequenceNumbers.getPartitionSequenceNumberMap()).thenReturn(ImmutableMap.of());
+ Mockito.when(sequenceNumbers.getStream()).thenReturn("test");
+
+ Mockito.when(task.getDataSchema()).thenReturn(schema);
+ Mockito.when(task.getIOConfig()).thenReturn(ioConfig);
+ Mockito.when(task.getTuningConfig()).thenReturn(tuningConfig);
+
+ TestasbleSeekableStreamIndexTaskRunner runner = new TestasbleSeekableStreamIndexTaskRunner(task, null, null, LockGranularity.TIME_CHUNK);
+
+ Mockito.when(row.getTimestamp()).thenReturn(now);
+ Assert.assertTrue(runner.withinMinMaxRecordTime(row));
+
+ Mockito.when(row.getTimestamp()).thenReturn(now.minusHours(2).minusMinutes(1));
+ Assert.assertFalse(runner.withinMinMaxRecordTime(row));
+
+ Mockito.when(row.getTimestamp()).thenReturn(now.plusHours(2).plusMinutes(1));
+ Assert.assertFalse(runner.withinMinMaxRecordTime(row));
+ }
+
+ @Test
+ public void testWithinMinMaxTimeNotPopulated()
+ {
+ DimensionsSpec dimensionsSpec = new DimensionsSpec(
+ Arrays.asList(
+ new StringDimensionSchema("d1"),
+ new StringDimensionSchema("d2")
+ )
+ );
+ DataSchema schema =
+ DataSchema.builder()
+ .withDataSource("datasource")
+ .withTimestamp(new TimestampSpec(null, null, null))
+ .withDimensions(dimensionsSpec)
+ .withGranularity(
+ new UniformGranularitySpec(Granularities.MINUTE, Granularities.NONE, null)
+ )
+ .build();
+
+ SeekableStreamIndexTaskTuningConfig tuningConfig = Mockito.mock(SeekableStreamIndexTaskTuningConfig.class);
+ SeekableStreamIndexTaskIOConfig<String, String> ioConfig = Mockito.mock(SeekableStreamIndexTaskIOConfig.class);
+ SeekableStreamStartSequenceNumbers<String, String> sequenceNumbers = Mockito.mock(SeekableStreamStartSequenceNumbers.class);
+ SeekableStreamEndSequenceNumbers<String, String> endSequenceNumbers = Mockito.mock(SeekableStreamEndSequenceNumbers.class);
+
+ DateTime now = DateTimes.nowUtc();
Review Comment:
## Unread local variable
Variable 'DateTime now' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/8449)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -137,7 +137,8 @@
spec.getIoConfig().getConsumerProperties(),
sortingMapper,
spec.getIoConfig().getConfigOverrides(),
- spec.getIoConfig().isMultiTopic()
+ spec.getIoConfig().isMultiTopic(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/8439)
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1829901071
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -246,6 +248,9 @@ public enum Status
private final Map<PartitionIdType, Long> partitionsThroughput = new HashMap<>();
+ private volatile Optional<DateTime> minMessageTime;
+ private volatile Optional<DateTime> maxMessageTime;
Review Comment:
just realized that it wouldn't overflow
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1841668533
##########
indexing-service/src/test/resources/log4j2.xml:
##########
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+
+<Configuration status="WARN">
+ <Loggers>
+ <Logger level="debug" name="org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner" additivity="false">
+ <AppenderRef ref="Console"/>
+ </Logger>
+ </Loggers>
+</Configuration>
Review Comment:
It atleast works locally for coverage.
--
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] supervisor: make rejection periods work with stopTasksCount (druid)
Posted by "adithyachakilam (via GitHub)" <gi...@apache.org>.
adithyachakilam commented on code in PR #17442:
URL: https://github.com/apache/druid/pull/17442#discussion_r1848680099
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -267,6 +271,18 @@ public SeekableStreamIndexTaskRunner(
this.ingestionState = IngestionState.NOT_STARTED;
this.lockGranularityToUse = lockGranularityToUse;
+ minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
+ maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
+
+ if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
+ Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d")
Review Comment:
https://github.com/apache/druid/pull/17490
--
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