-
Notifications
You must be signed in to change notification settings - Fork 202
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into enh/4838-support-plugin-loading-in-conifg
Signed-off-by: George Chen <qchea@amazon.com>
- Loading branch information
Showing
87 changed files
with
5,155 additions
and
300 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
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.