-
Notifications
You must be signed in to change notification settings - Fork 190
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'opensearch-project:main' into saas-connector-sdk
- Loading branch information
Showing
90 changed files
with
3,655 additions
and
609 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
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
81 changes: 81 additions & 0 deletions
81
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,81 @@ | ||
| /* | ||
| * 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 org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| 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 static final Logger LOG = LoggerFactory.getLogger(PipelineShutdown.class); | ||
|
|
||
| private final AtomicBoolean stopRequested = new AtomicBoolean(false); | ||
| private final Duration pipelineConfiguredBufferDrainTimeout; | ||
| private final Clock clock; | ||
| private final String pipelineName; | ||
| private Instant shutdownRequestedAt; | ||
| private Instant forceStopReadingBuffersAt; | ||
| private Duration bufferDrainTimeoutOverride; | ||
| private Duration bufferDrainTimeout; | ||
|
|
||
| PipelineShutdown(final String pipelineName, final Buffer<?> buffer) { | ||
| this(pipelineName, buffer, Clock.systemDefaultZone()); | ||
| } | ||
|
|
||
| PipelineShutdown(String pipelineName, final Buffer<?> buffer, final Clock clock) { | ||
| this.pipelineName = pipelineName; | ||
| pipelineConfiguredBufferDrainTimeout = Objects.requireNonNull(buffer.getDrainTimeout()); | ||
| bufferDrainTimeout = pipelineConfiguredBufferDrainTimeout; | ||
| 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; | ||
| bufferDrainTimeout = bufferDrainTimeoutOverride; | ||
| } | ||
|
|
||
| LOG.info("Started shutdown for pipeline {}. Requested at {}. Force stop reading buffers at {}. The buffer drain timeout to use is {}", | ||
| pipelineName, shutdownRequestedAt, forceStopReadingBuffersAt, bufferDrainTimeout); | ||
| } | ||
|
|
||
| boolean isStopRequested() { | ||
| return stopRequested.get(); | ||
| } | ||
|
|
||
| boolean isForceStopReadingBuffers() { | ||
| return forceStopReadingBuffersAt != null && now().isAfter(forceStopReadingBuffersAt); | ||
| } | ||
|
|
||
| public Duration getBufferDrainTimeout() { | ||
| return 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.