Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

StreamingTaskRunner: Close the rejection period updater executor service (#17490)

This commit is contained in:
Adithya Chakilam 2024-11-19 14:49:20 -06:00 committed by GitHub
parent 8853c7e5c6
commit c1d6328249
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
View File

@ -128,6 +128,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@ -249,6 +250,7 @@ public abstract class SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
private volatile DateTime minMessageTime; private volatile DateTime minMessageTime;
private volatile DateTime maxMessageTime; private volatile DateTime maxMessageTime;
private final ScheduledExecutorService rejectionPeriodUpdaterExec;
public SeekableStreamIndexTaskRunner( public SeekableStreamIndexTaskRunner(
final SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType, RecordType> task, final SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType, RecordType> task,
@ -273,15 +275,15 @@ public abstract class SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN); minMessageTime = ioConfig.getMinimumMessageTime().or(DateTimes.MIN);
maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX); maxMessageTime = ioConfig.getMaximumMessageTime().or(DateTimes.MAX);
rejectionPeriodUpdaterExec = Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d");
if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) { if (ioConfig.getRefreshRejectionPeriodsInMinutes() != null) {
Execs.scheduledSingleThreaded("RejectionPeriodUpdater-Exec--%d") rejectionPeriodUpdaterExec
.scheduleWithFixedDelay( .scheduleWithFixedDelay(
this::refreshMinMaxMessageTime, this::refreshMinMaxMessageTime,
ioConfig.getRefreshRejectionPeriodsInMinutes(), ioConfig.getRefreshRejectionPeriodsInMinutes(),
ioConfig.getRefreshRejectionPeriodsInMinutes(), ioConfig.getRefreshRejectionPeriodsInMinutes(),
TimeUnit.MINUTES TimeUnit.MINUTES);
);
} }
resetNextCheckpointTime(); resetNextCheckpointTime();
} }
@ -940,6 +942,7 @@ public abstract class SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
toolbox.getDruidNodeAnnouncer().unannounce(discoveryDruidNode); toolbox.getDruidNodeAnnouncer().unannounce(discoveryDruidNode);
toolbox.getDataSegmentServerAnnouncer().unannounce(); toolbox.getDataSegmentServerAnnouncer().unannounce();
} }
rejectionPeriodUpdaterExec.shutdown();
} }
catch (Throwable e) { catch (Throwable e) {
if (caughtExceptionOuter != null) { if (caughtExceptionOuter != null) {