Support AWS Kinesis Data Streams as a Source #4836

sb2k16 · 2024-08-14T19:01:58Z

Description

Implementation for a source plugin in data-prepper to support ingestion of Kinesis Data Streams. It uses KCL (Kinesls Client Library) under the hood for ingestion purposes.

KCL features used:

multi-stream support
lease coordination using a single dynamoDB table per pipeline
enhanced fan-out reads

Pipeline configuration for source:

  source:
    kinesis:
      streams:
        - stream_name: "stream-a"
        - stream_name: "stream-b"
          initial_position: "TRIM_HORIZON" # Default = "LATEST"
      aws:
          ...
      acknowledgments: true
      consumer_strategy: "fan-out" or "polling" # Default = "fan-out"
      codec:
        ndjson:

Issues Resolved

Resolves #1082

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[X ] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

...rg/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseCoordinationTableConfig.java

kkondaka · 2024-08-21T02:22:27Z

...is-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/ClientFactory.java

@@ -0,0 +1,50 @@
+package org.opensearch.dataprepper.plugins.kinesis.source;


I know we aren't following it consistently but I think it is better to name this as KinesisClientFactory.java. There are too many "ClientFactory.java" and it becomes search more difficult.

...org/opensearch/dataprepper/plugins/kinesis/source/configuration/AwsAuthenticationConfig.java

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

data-prepper-plugins/kinesis-source/src/test/resources/simple_pipeline_with_extensions.yaml

dlvenable

@sb2k16 , Thank you for this contribution. This initial PR review has a number of high-level comments.

data-prepper-plugins/kinesis-source/build.gradle

.../org/opensearch/dataprepper/plugins/kinesis/extension/DefaultKinesisLeaseConfigSupplier.java

data-prepper-plugins/kinesis-source/build.gradle

dlvenable · 2024-08-28T22:18:28Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+        }
+    }
+
+    private void processRecord(KinesisClientRecord record, Consumer<Record<Event>> eventConsumer) throws IOException {


I think we should split the logic for processing an individual record out of this class to make the class easier to extend and test over time.

@sb2k16 , Let's make this a separate class. Perhaps KinesisRecordConverter. It can be responsible for all the work for taking the record and using the codec. You can have some good tests on this as well.

We've had difficulty expanding our tests in other similar locations because we combine too many things in a single class.

dlvenable · 2024-08-28T22:19:08Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+
+            // Checkpoint for shard
+            if (kinesisStreamConfig.isEnableCheckPoint() && System.currentTimeMillis() - lastCheckpointTimeInMillis > checkpointIntervalMilliSeconds) {
+                LOG.info("Regular checkpointing for shard " + kinesisShardId);


Suggested change

LOG.info("Regular checkpointing for shard " + kinesisShardId);

LOG.debug("Regular checkpointing for shard {}", kinesisShardId);

Use Sl4fj string interpolation.

Also, this is probably more of a debug log.

Thanks @dlvenable. I have fixed this.

Updating the checkpoint synchronously is a pattern that we should move away from, as it can lead to this issue (#4422). You can create a separate thread to do this checkpointing. To handle it for acknowledgments, you can add a progress check (example here: #4918). Unless KCL handles the ownership aspect and the checkpointing is just to show how far it has gone in a shard

@sbose2k21 , I believe you missed this one.

...s-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java

graytaylor0 · 2024-09-10T19:28:27Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisClientFactory.java

+        this.awsAuthenticationConfig = awsAuthenticationConfig;
+    }
+
+    public DynamoDbAsyncClient buildDynamoDBClient(Region region) {


Is it recommended to use the Async client? If so why?

Thanks @graytaylor0 for your review. This is required to construct ConfigsBuilder class in KCL

graytaylor0 · 2024-09-10T19:29:23Z

...s-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java

+        }
+
+        if (sourceConfig.getStreams() == null || sourceConfig.getStreams().isEmpty()) {
+            throw new IllegalStateException("Streams are empty!");


We should throw InvalidPluginConfiugrationException here

Let's also make a clearer and less excited message.

No Kinesis streams provided.

graytaylor0 · 2024-09-10T20:01:12Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+
+            // Checkpoint for shard
+            if (kinesisStreamConfig.isEnableCheckPoint() && System.currentTimeMillis() - lastCheckpointTimeInMillis > checkpointIntervalMilliSeconds) {
+                LOG.info("Regular checkpointing for shard " + kinesisShardId);


Updating the checkpoint synchronously is a pattern that we should move away from, as it can lead to this issue (#4422). You can create a separate thread to do this checkpointing. To handle it for acknowledgments, you can add a progress check (example here: #4918). Unless KCL handles the ownership aspect and the checkpointing is just to show how far it has gone in a shard

dlvenable

Thank you @sb2k16 for this change! There are a few outstanding comments from the last review. And I have a few other comments as well.

dlvenable · 2024-09-10T21:21:48Z

...ava/org/opensearch/dataprepper/plugins/kinesis/source/configuration/KinesisStreamConfig.java

+
+    @Getter
+    @JsonProperty("enable_checkpoint")
+    private boolean enableCheckPoint = DEFAULT_ENABLE_CHECKPOINT;


Why would somebody disable checkpointing? Also, the default is false here. Why is that?

Since, we will be enabling acknowledgements, this is set to false by default.

dlvenable · 2024-09-10T21:23:05Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+            // Checkpoint for shard
+            if (kinesisStreamConfig.isEnableCheckPoint() && System.currentTimeMillis() - lastCheckpointTimeInMillis > checkpointInterval.toMillis()) {
+                LOG.info("Regular checkpointing for shard " + kinesisShardId);
+                checkpoint(processRecordsInput.checkpointer());


I believe we should not call this when end-to-end acknowledgements are enabled. Otherwise, we will checkpoint prematurely.

dlvenable · 2024-09-10T21:25:38Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+        return acknowledgementSetManager.create((result) -> {
+            acknowledgementSetCallbackCounter.increment();
+            if (result) {
+                LOG.info("acknowledgements received");


This should be a debug level log.

dlvenable · 2024-09-10T21:25:41Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+                LOG.info("acknowledgements received");
+                checkpoint(processRecordsInput.checkpointer());
+            } else {
+                LOG.info("acknowledgements received with false");


This should be a debug level log.

dlvenable · 2024-09-10T21:28:07Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+        this.kinesisStreamConfig = getStreamConfig(kinesisSourceConfig);
+        final PluginModel codecConfiguration = kinesisSourceConfig.getCodec();
+        final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(), codecConfiguration.getPluginSettings());
+        this.codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings);


Might KCL create multiple of these processors? We can create the InputCodec once, either in KinesisService (probably the ideal location) or KinesisSharedRecordProcessorFactory. Then pass it in here. This way, multiple instances of KinesisRecordProcessor all share one instance of InputCodec.

This also prepares us to use #4838 once implemented.

Thanks @dlvenable. I have passed the KinesisRecordConverter object to each KinesisRecordProcessor which is instantiated in KinesisSharedRecordProcessorFactory.

dlvenable · 2024-09-10T21:37:54Z

.../org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessorTest.java

+        kinesisRecordProcessor = new KinesisRecordProcessor(buffer, kinesisSourceConfig, acknowledgementSetManager, pluginMetrics, pluginFactory, streamIdentifier);
+        kinesisRecordProcessor.initialize(initializationInput);
+
+        Thread.sleep(2000);


Use Awaitility instead of a sleep.

I have modified the test to remove this.

dlvenable · 2024-09-10T21:37:59Z

.../org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessorTest.java

+        kinesisRecordProcessor = new KinesisRecordProcessor(buffer, kinesisSourceConfig, acknowledgementSetManager, pluginMetrics, pluginFactory, streamIdentifier);
+        kinesisRecordProcessor.initialize(initializationInput);
+
+        Thread.sleep(2000);


Use Awaitility instead of a sleep.

I have modified the test to remove this.

dlvenable · 2024-09-10T21:38:03Z

.../org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessorTest.java

+        kinesisRecordProcessor = new KinesisRecordProcessor(buffer, kinesisSourceConfig, acknowledgementSetManager, pluginMetrics, pluginFactory, streamIdentifier);
+        kinesisRecordProcessor.initialize(initializationInput);
+
+        Thread.sleep(2000);


Use Awaitility instead of a sleep.

I have modified the test to remove this.

dlvenable · 2024-09-10T21:38:08Z

.../org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessorTest.java

+        kinesisRecordProcessor = new KinesisRecordProcessor(buffer, kinesisSourceConfig, acknowledgementSetManager, pluginMetrics, pluginFactory, streamIdentifier);
+        kinesisRecordProcessor.initialize(initializationInput);
+
+        Thread.sleep(2000);


Use Awaitility instead of a sleep.

I have modified the test to remove this.

dlvenable · 2024-09-10T21:40:23Z

...e/src/main/java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfig.java

@@ -0,0 +1,10 @@
+package org.opensearch.dataprepper.plugins.kinesis.extension;


Every class needs a license header. Please add to all files.

See: https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md#license-headers

Hint: If you use IntelliJ you can configure it to automatically add this. And you can have it add the headers to all the files in this Gradle module (since they are all new).

Signed-off-by: Souvik Bose <souvbose@amazon.com>

kkondaka · 2024-09-14T06:08:44Z

...ava/org/opensearch/dataprepper/plugins/kinesis/source/configuration/KinesisSourceConfig.java

+
+    @JsonProperty("acknowledgments")
+    @Getter
+    private boolean acknowledgments = false;


Should we start making acknowledgements as true by default to all pull sources? If yes, we should start with this. @dlvenable

I think we should keep the behavior consistent for now. I do think we should revisit in Data Prepper 3.0. It may make sense to have a disable_acknowledgements: true in that case. But, let's do it consistently then.

dlvenable · 2024-09-17T16:08:30Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

        this.recordProcessingErrors = pluginMetrics.counterWithTags(KINESIS_RECORD_PROCESSING_ERRORS, KINESIS_STREAM_TAG_KEY, streamIdentifier.streamName());
        this.checkpointFailures = pluginMetrics.counterWithTags(KINESIS_CHECKPOINT_FAILURES, KINESIS_STREAM_TAG_KEY, streamIdentifier.streamName());
        this.checkpointInterval = kinesisStreamConfig.getCheckPointInterval();
-        this.buffer = buffer;
+        this.bufferAccumulator = bufferAccumulator;
+        this.kinesisCheckpointerTracker = new KinesisCheckpointerTracker();


Rather than include setKinesisCheckpointTracker, just pass this in the constructor. Then you can provide the mocks in the test using that. You can then remove the setter as this can lead to code bugs down the line. And you can make the field final.

Thanks @dlvenable . I have removed the setter method and passed in the checkpoint tracker through the constructor.

dlvenable · 2024-09-17T16:10:40Z

.../org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisCheckpointerTracker.java

+        checkpointerRecordList.get(extendedSequenceNumber).setReadyToCheckpoint(true);
+    }
+
+    public synchronized Optional<KinesisCheckpointerRecord> getLatestAvailableCheckpointRecord() {


Maybe rename this to popLatestAvailableCheckpointRecord or something similar. It is mutating the structure and a get method doesn't convey that.

I have renamed this method to popLatestReadyToCheckpointRecord

dlvenable · 2024-09-17T16:14:39Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+
+                Optional<KinesisCheckpointerRecord> kinesisCheckpointerRecordOptional = kinesisCheckpointerTracker.getLatestAvailableCheckpointRecord();
+                if (kinesisCheckpointerRecordOptional.isPresent()) {
+                    RecordProcessorCheckpointer recordProcessorCheckpointer = kinesisCheckpointerRecordOptional.get().getCheckpointer();


Can you combine these lines with 201-205?

Thanks @dlvenable. I have refactored these lines and added a new method doCheckpoint.

dlvenable

Thank you @sb2k16 for this great contribution!

graytaylor0 · 2024-09-18T22:05:47Z

...c/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisMultiStreamTracker.java

+            try {
+                streamConfig = getStreamConfig(kinesisStreamConfig);
+            } catch (Exception e) {
+                throw new RuntimeException(e);


What kind of exceptions get thrown here? If it's a config error we should throw InvalidPluginConfigurationException and provide context with the message where we can

This is an internal StreamConfig class which is passed to KCL. It is not data prepper kinesis stream config.

graytaylor0 · 2024-09-18T22:06:18Z

...c/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisMultiStreamTracker.java

+    }
+
+    private String getStreamIdentifierString(StreamDescription streamDescription) {
+        String accountId = streamDescription.streamARN().split(COLON)[4];


Is this an actual ARN, or just an identifier? If Arn we should use the Arn class to get the account id

Thanks @graytaylor0. I will fix this.

graytaylor0 · 2024-09-18T22:07:18Z

...c/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisMultiStreamTracker.java

+        return new FormerStreamsLeasesDeletionStrategy.AutoDetectionAndDeferredDeletionStrategy() {
+            @Override
+            public Duration waitPeriodToDeleteFormerStreams() {
+                return Duration.ofSeconds(10);


What are trade offs for having this higher or lower?

It is the strategy to perform cleaning up of the leases for streams that just has been deleted. The wait period is to allow for any pending processing of records before cleaning up the lease from the database.

graytaylor0 · 2024-09-18T22:08:23Z

...s-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java

+        this.applicationName = pipelineName;
+        this.workerIdentifierGenerator = workerIdentifierGenerator;
+        this.executorService = Executors.newFixedThreadPool(1);
+        final PluginModel codecConfiguration = kinesisSourceConfig.getCodec();


Is this codec the type of data they send to Kinesis? It's not just JSON records from Kinesis streams? Or is it always provided from Kinesis as just bytes?

When we read from Kinesis, it always comes in as bytes. Reference Record

graytaylor0 · 2024-09-18T22:13:28Z

...java/org/opensearch/dataprepper/plugins/kinesis/source/processor/KinesisRecordProcessor.java

+
+    @Override
+    public void leaseLost(LeaseLostInput leaseLostInput) {
+        LOG.debug("Lease Lost");


Is this worth having a metric for or does kinesis already track it? Why does this happen?

KCL already tracks this through a metric. This can happen if the current worker holding the lease is evicted by another worker which acquires the lease for the specific shard.

...org/opensearch/dataprepper/plugins/source/kinesis/configuration/AwsAuthenticationConfig.java

...java/org/opensearch/dataprepper/plugins/source/kinesis/processor/KinesisRecordProcessor.java

Signed-off-by: Souvik Bose <souvbose@amazon.com>

sb2k16 requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, KarstenSchnitter, dlvenable and oeyh as code owners August 14, 2024 19:01

sb2k16 force-pushed the kinesis-code-checkin branch from 81c887c to 93ababc Compare August 21, 2024 01:16

kkondaka reviewed Aug 21, 2024

View reviewed changes

sb2k16 force-pushed the kinesis-code-checkin branch from 1ca0077 to 09212fa Compare August 28, 2024 16:52

dlvenable requested changes Aug 28, 2024

View reviewed changes

sb2k16 force-pushed the kinesis-code-checkin branch 3 times, most recently from 7da25c7 to ba2532d Compare August 29, 2024 18:45

graytaylor0 reviewed Sep 10, 2024

View reviewed changes

dlvenable requested changes Sep 10, 2024

View reviewed changes

sbose2k21 added 6 commits September 10, 2024 21:33

Kinesis data streams ingestion functionality checkin

c2dcf76

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Add support for extensions to configure lease coordination table.

1523b8a

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Cleanup unused code for metrics

f8f89fb

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Update tests and changes per review comments

0567858

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Address review comments

63a2f91

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Add license header to all the classes

0fd763b

Signed-off-by: Souvik Bose <souvbose@amazon.com>

sb2k16 force-pushed the kinesis-code-checkin branch 3 times, most recently from 82aa510 to 57f5d42 Compare September 13, 2024 16:35

Address review comments

92a797b

Signed-off-by: Souvik Bose <souvbose@amazon.com>

sb2k16 force-pushed the kinesis-code-checkin branch from 57f5d42 to 92a797b Compare September 13, 2024 23:34

kkondaka reviewed Sep 14, 2024

View reviewed changes

dlvenable requested changes Sep 17, 2024

View reviewed changes

dlvenable previously approved these changes Sep 18, 2024

View reviewed changes

graytaylor0 reviewed Sep 18, 2024

View reviewed changes

Rename function and address comments.

806f78d

Signed-off-by: Souvik Bose <souvbose@amazon.com>

sb2k16 dismissed dlvenable’s stale review via 806f78d September 19, 2024 00:39

sb2k16 force-pushed the kinesis-code-checkin branch from 4ded035 to 806f78d Compare September 19, 2024 00:39

graytaylor0 approved these changes Sep 19, 2024

View reviewed changes

dlvenable approved these changes Sep 19, 2024

View reviewed changes

dlvenable merged commit a3bd538 into opensearch-project:main Sep 19, 2024
71 of 74 checks passed

		@@ -0,0 +1,50 @@
		package org.opensearch.dataprepper.plugins.kinesis.source;

	LOG.info("Regular checkpointing for shard " + kinesisShardId);
	LOG.debug("Regular checkpointing for shard {}", kinesisShardId);

		@@ -0,0 +1,10 @@
		package org.opensearch.dataprepper.plugins.kinesis.extension;

Support AWS Kinesis Data Streams as a Source #4836

Support AWS Kinesis Data Streams as a Source #4836

Conversation

sb2k16 commented Aug 14, 2024 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sb2k16 Sep 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sb2k16 commented Aug 14, 2024 •

edited

Loading

sb2k16 Sep 10, 2024 •

edited

Loading