-
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.
Add support for extensions to configure lease coordination table.
Signed-off-by: Souvik Bose <[email protected]>
- Loading branch information
Showing
34 changed files
with
425 additions
and
67 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
17 changes: 17 additions & 0 deletions
17
...g/opensearch/dataprepper/plugins/kinesis/extension/DefaultKinesisLeaseConfigSupplier.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,17 @@ | ||
| package org.opensearch.dataprepper.plugins.kinesis.extension; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| public class DefaultKinesisLeaseConfigSupplier implements KinesisLeaseConfigSupplier { | ||
|
|
||
| private KinesisLeaseConfig kinesisLeaseConfig; | ||
|
|
||
| public DefaultKinesisLeaseConfigSupplier(final KinesisLeaseConfig kinesisLeaseConfig) { | ||
| this.kinesisLeaseConfig = kinesisLeaseConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<KinesisLeaseConfig> getKinesisExtensionLeaseConfig() { | ||
| return kinesisLeaseConfig != null ? Optional.of(kinesisLeaseConfig) : Optional.empty(); | ||
| } | ||
| } |
10 changes: 10 additions & 0 deletions
10
...rc/main/java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfig.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,10 @@ | ||
| package org.opensearch.dataprepper.plugins.kinesis.extension; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import lombok.Getter; | ||
|
|
||
| @Getter | ||
| public class KinesisLeaseConfig { | ||
| @JsonProperty("lease_coordination_table") | ||
| private String leaseCoordinationTable; | ||
| } |
21 changes: 21 additions & 0 deletions
21
...ava/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfigExtension.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,21 @@ | ||
| package org.opensearch.dataprepper.plugins.kinesis.extension; | ||
|
|
||
| import org.opensearch.dataprepper.model.annotations.DataPrepperExtensionPlugin; | ||
| import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor; | ||
| import org.opensearch.dataprepper.model.plugin.ExtensionPlugin; | ||
| import org.opensearch.dataprepper.model.plugin.ExtensionPoints; | ||
|
|
||
| @DataPrepperExtensionPlugin(modelType = KinesisLeaseConfig.class, rootKeyJsonPath = "/kinesis_lease_config", allowInPipelineConfigurations = true) | ||
| public class KinesisLeaseConfigExtension implements ExtensionPlugin { | ||
|
|
||
| private KinesisLeaseConfigSupplier kinesisLeaseConfigSupplier; | ||
| @DataPrepperPluginConstructor | ||
| public KinesisLeaseConfigExtension(final KinesisLeaseConfig kinesisLeaseConfig) { | ||
| this.kinesisLeaseConfigSupplier = new DefaultKinesisLeaseConfigSupplier(kinesisLeaseConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public void apply(final ExtensionPoints extensionPoints) { | ||
| extensionPoints.addExtensionProvider(new KinesisLeaseConfigProvider(this.kinesisLeaseConfigSupplier)); | ||
| } | ||
| } |
23 changes: 23 additions & 0 deletions
23
...java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfigProvider.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,23 @@ | ||
| package org.opensearch.dataprepper.plugins.kinesis.extension; | ||
|
|
||
| import org.opensearch.dataprepper.model.plugin.ExtensionProvider; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| class KinesisLeaseConfigProvider implements ExtensionProvider<KinesisLeaseConfigSupplier> { | ||
| private final KinesisLeaseConfigSupplier kinesisLeaseConfigSupplier; | ||
|
|
||
| public KinesisLeaseConfigProvider(final KinesisLeaseConfigSupplier kinesisLeaseConfigSupplier) { | ||
| this.kinesisLeaseConfigSupplier = kinesisLeaseConfigSupplier; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<KinesisLeaseConfigSupplier> provideInstance(Context context) { | ||
| return Optional.of(this.kinesisLeaseConfigSupplier); | ||
| } | ||
|
|
||
| @Override | ||
| public Class<KinesisLeaseConfigSupplier> supportedClass() { | ||
| return KinesisLeaseConfigSupplier.class; | ||
| } | ||
| } |
10 changes: 10 additions & 0 deletions
10
...java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfigSupplier.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,10 @@ | ||
| package org.opensearch.dataprepper.plugins.kinesis.extension; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| public interface KinesisLeaseConfigSupplier { | ||
|
|
||
| default Optional<KinesisLeaseConfig> getKinesisExtensionLeaseConfig() { | ||
| return Optional.empty(); | ||
| } | ||
| } |
4 changes: 2 additions & 2 deletions
4
...plugins/source/kinesis/ClientFactory.java → ...plugins/kinesis/source/ClientFactory.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
6 changes: 3 additions & 3 deletions
6
...ce/kinesis/KinesisMultiStreamTracker.java → ...sis/source/KinesisMultiStreamTracker.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
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
2 changes: 1 addition & 1 deletion
2
...ce/kinesis/WorkerIdentifierGenerator.java → ...sis/source/WorkerIdentifierGenerator.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
2 changes: 1 addition & 1 deletion
2
...onfiguration/AwsAuthenticationConfig.java → ...onfiguration/AwsAuthenticationConfig.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
2 changes: 1 addition & 1 deletion
2
...nesis/configuration/ConsumerStrategy.java → ...ource/configuration/ConsumerStrategy.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
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
2 changes: 1 addition & 1 deletion
2
...is/configuration/KinesisStreamConfig.java → ...ce/configuration/KinesisStreamConfig.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
2 changes: 1 addition & 1 deletion
2
...iguration/KinesisStreamPollingConfig.java → ...iguration/KinesisStreamPollingConfig.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
38 changes: 38 additions & 0 deletions
38
...g/opensearch/dataprepper/plugins/kinesis/source/metrics/MicrometerFilterMetricsScope.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,38 @@ | ||
| //package org.opensearch.dataprepper.plugins.kinesis.source.metrics; | ||
| // | ||
| //import com.google.common.collect.ImmutableSet; | ||
| //import org.opensearch.dataprepper.metrics.PluginMetrics; | ||
| //import org.slf4j.Logger; | ||
| //import org.slf4j.LoggerFactory; | ||
| //import software.amazon.awssdk.services.cloudwatch.model.StandardUnit; | ||
| //import software.amazon.kinesis.metrics.FilteringMetricsScope; | ||
| //import software.amazon.kinesis.metrics.MetricsLevel; | ||
| // | ||
| //public class MicrometerFilterMetricsScope extends FilteringMetricsScope { | ||
| // | ||
| // private static final Logger LOG = LoggerFactory.getLogger(MicrometerFilterMetricsScope.class); | ||
| // | ||
| // private final PluginMetrics pluginMetrics; | ||
| // | ||
| // public MicrometerFilterMetricsScope(final PluginMetrics pluginMetrics) { | ||
| // super(MetricsLevel.SUMMARY, ImmutableSet.of(METRICS_DIMENSIONS_ALL)); | ||
| // this.pluginMetrics = pluginMetrics; | ||
| // } | ||
| // | ||
| // @Override | ||
| // public void addData(String name, double value, StandardUnit unit) { | ||
| // // TODO: report metrics. | ||
| // LOG.info("{} {} {}", name, value, unit); | ||
| // } | ||
| // | ||
| // @Override | ||
| // public void addData(String name, double value, StandardUnit unit, MetricsLevel level) { | ||
| // // TODO: report metrics. | ||
| // LOG.info("{} {} {} {}", name, value, unit, level); | ||
| // } | ||
| // | ||
| // @Override | ||
| // public void end() { | ||
| // // TODO: Update this | ||
| // } | ||
| //} |
18 changes: 18 additions & 0 deletions
18
...va/org/opensearch/dataprepper/plugins/kinesis/source/metrics/MicrometerMetricFactory.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,18 @@ | ||
| //package org.opensearch.dataprepper.plugins.kinesis.source.metrics; | ||
| // | ||
| //import org.opensearch.dataprepper.metrics.PluginMetrics; | ||
| //import software.amazon.kinesis.metrics.MetricsFactory; | ||
| //import software.amazon.kinesis.metrics.MetricsScope; | ||
| // | ||
| //public class MicrometerMetricFactory implements MetricsFactory { | ||
| // | ||
| // final PluginMetrics pluginMetrics; | ||
| // | ||
| // public MicrometerMetricFactory(PluginMetrics pluginMetrics) { | ||
| // this.pluginMetrics = pluginMetrics; | ||
| // } | ||
| // | ||
| // public MetricsScope createMetrics() { | ||
| // return new MicrometerFilterMetricsScope(pluginMetrics); | ||
| // } | ||
| //} |
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
4 changes: 2 additions & 2 deletions
4
...r/KinesisShardRecordProcessorFactory.java → ...r/KinesisShardRecordProcessorFactory.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
Oops, something went wrong.