-
Notifications
You must be signed in to change notification settings - Fork 188
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Variable drain timeouts when shutting down over HTTP shutdown. (#4970)
Variable drain timeouts when shutting down over HTTP shutdown. Adds two new parameters to the shutdown API. The first is bufferReadTimeout which controls the amount of time to wait for the buffer to be empty. The second is bufferDrainTimeout which controls the overall wait time for the process worker threads to complete. To support Data Prepper durations in HTTP query parameters, I extracted the parsing logic for durations out of DataPrepperDurationDeserializer and into a new DataPrepperDurationParser class. Resolves #4966. Signed-off-by: David Venable <dlv@amazon.com>
- Loading branch information
Showing
14 changed files
with
751 additions
and
121 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
data-prepper-core/src/main/java/org/opensearch/dataprepper/DataPrepperShutdownOptions.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper; | ||
|
|
||
| import java.time.Duration; | ||
|
|
||
| public class DataPrepperShutdownOptions { | ||
| private final Duration bufferReadTimeout; | ||
| private final Duration bufferDrainTimeout; | ||
|
|
||
| public static DataPrepperShutdownOptions defaultOptions() { | ||
| return new DataPrepperShutdownOptions(builder()); | ||
| } | ||
|
|
||
| public static Builder builder() { | ||
| return new Builder(); | ||
| } | ||
|
|
||
| public static class Builder { | ||
| private Duration bufferReadTimeout; | ||
| private Duration bufferDrainTimeout; | ||
|
|
||
| private Builder() { | ||
| } | ||
|
|
||
| public Builder withBufferReadTimeout(final Duration bufferReadTimeout) { | ||
| this.bufferReadTimeout = bufferReadTimeout; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder withBufferDrainTimeout(final Duration bufferDrainTimeout) { | ||
| this.bufferDrainTimeout = bufferDrainTimeout; | ||
| return this; | ||
| } | ||
|
|
||
| public DataPrepperShutdownOptions build() { | ||
| return new DataPrepperShutdownOptions(this); | ||
| } | ||
| } | ||
|
|
||
| private DataPrepperShutdownOptions(final Builder builder) { | ||
| this.bufferReadTimeout = builder.bufferReadTimeout; | ||
| this.bufferDrainTimeout = builder.bufferDrainTimeout; | ||
|
|
||
| if(bufferReadTimeout != null && bufferDrainTimeout != null) { | ||
| if (bufferReadTimeout.compareTo(bufferDrainTimeout) > 0) { | ||
| throw new IllegalArgumentException("Buffer read timeout cannot be greater than buffer drain timeout"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public Duration getBufferReadTimeout() { | ||
| return bufferReadTimeout; | ||
| } | ||
|
|
||
| public Duration getBufferDrainTimeout() { | ||
| return bufferDrainTimeout; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
70 changes: 70 additions & 0 deletions
70
data-prepper-core/src/main/java/org/opensearch/dataprepper/pipeline/PipelineShutdown.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.pipeline; | ||
|
|
||
| import org.opensearch.dataprepper.DataPrepperShutdownOptions; | ||
| import org.opensearch.dataprepper.model.buffer.Buffer; | ||
|
|
||
| import java.time.Clock; | ||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.util.Objects; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| class PipelineShutdown { | ||
| private final AtomicBoolean stopRequested = new AtomicBoolean(false); | ||
| private final Duration bufferDrainTimeout; | ||
| private final Clock clock; | ||
| private Instant shutdownRequestedAt; | ||
| private Instant forceStopReadingBuffersAt; | ||
| private Duration bufferDrainTimeoutOverride; | ||
|
|
||
| PipelineShutdown(final Buffer<?> buffer) { | ||
| this(buffer, Clock.systemDefaultZone()); | ||
| } | ||
|
|
||
| PipelineShutdown(final Buffer<?> buffer, final Clock clock) { | ||
| bufferDrainTimeout = Objects.requireNonNull(buffer.getDrainTimeout()); | ||
| this.clock = clock; | ||
| } | ||
|
|
||
| public void shutdown(final DataPrepperShutdownOptions dataPrepperShutdownOptions) { | ||
| final boolean stopPreviouslyRequested = stopRequested.get(); | ||
| if(stopPreviouslyRequested) { | ||
| return; | ||
| } | ||
|
|
||
| stopRequested.set(true); | ||
| shutdownRequestedAt = now(); | ||
|
|
||
| final Duration bufferReadTimeout = dataPrepperShutdownOptions.getBufferReadTimeout(); | ||
| if(bufferReadTimeout != null) { | ||
| forceStopReadingBuffersAt = shutdownRequestedAt.plus(bufferReadTimeout); | ||
| } | ||
|
|
||
| final Duration bufferDrainTimeoutOverride = dataPrepperShutdownOptions.getBufferDrainTimeout(); | ||
| if(bufferDrainTimeoutOverride != null) { | ||
| this.bufferDrainTimeoutOverride = bufferDrainTimeoutOverride; | ||
| } | ||
| } | ||
|
|
||
| boolean isStopRequested() { | ||
| return stopRequested.get(); | ||
| } | ||
|
|
||
| boolean isForceStopReadingBuffers() { | ||
| return forceStopReadingBuffersAt != null && now().isAfter(forceStopReadingBuffersAt); | ||
| } | ||
|
|
||
| public Duration getBufferDrainTimeout() { | ||
| return bufferDrainTimeoutOverride != null ? | ||
| bufferDrainTimeoutOverride : bufferDrainTimeout; | ||
| } | ||
|
|
||
| private Instant now() { | ||
| return Instant.ofEpochMilli(clock.millis()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.