Kinesis ingestion with empty shards #12792

AmatyaAvadhanula · 2022-07-15T09:06:55Z

Enables Kinesis ingestion with empty shards

Description

Kinesis ingestion requires all shards to have at least 1 record at the required position in druid.
Even if this is satisified initially, resharding the stream can lead to empty intermediate shards. A significant delay in writing to newly created shards was also problematic.

Kinesis shard sequence numbers are big integers. Introduce two more custom sequence tokens UNREAD_TRIM_HORIZON and UNREAD_LATEST to indicate that a shard has not been read from and that it needs to be read from the start or the end respectively.
These values can be used to avoid the need to read at least one record to obtain a sequence number for ingesting a newly discovered shard.

If a record cannot be obtained immediately, use a marker to obtain the relevant shardIterator and use this shardIterator to obtain a valid sequence number. As long as a valid sequence number is not obtained, continue storing the token as the offset.

These tokens (UNREAD_TRIM_HORIZON and UNREAD_LATEST) are logically ordered to be earlier than any valid sequence number.

However, the ordering requires a few subtle changes to the existing mechanism for record sequence validation:

The sequence availability check ensures that the current offset is before the earliest available sequence in the shard. However, current token being an UNREAD token indicates that any sequence number in the shard is valid (despite the ordering)
Kinesis sequence numbers are inclusive i.e if current sequence == end sequence, there are more records left to read.
However, the equality check is exclusive when dealing with UNREAD tokens.

Key changed/added classes in this PR

KinesisSequenceNumber
KinesisRecordSupplier

This PR has:

been self-reviewed.
- using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
[] added documentation for new or modified features or behaviors.
[] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added or updated version, license, or notice information in licenses.yaml
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
added integration tests.
been tested in a test Druid cluster.

maytasm · 2022-07-27T05:17:24Z

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java

+  }
+
+  @Override
+  public boolean isAvailableWithEarliest(OrderedSequenceNumber<String> earliest)


The method name is a bit hard to understand. Maybe javadoc will help

Thanks for the feedback. Will add one

maytasm · 2022-07-27T05:17:30Z

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java

+  }
+
+  @Override
+  public boolean isMoreToReadBeforeReadingRecord(OrderedSequenceNumber<String> end)


The method name is a bit hard to understand. Maybe javadoc will help

This method is based on an existing method of the same name. I'll add a javadoc though

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java

maytasm

Can you also add some Integration test (See integration test group "kinesis-index")?

AmatyaAvadhanula · 2022-07-27T06:12:02Z

@maytasm, thanks for the review.

Can you also add some Integration test (See integration test group "kinesis-index")?

Sure, I can add a simple test where the records are queried after ingesting from a stream with at least one empty shard.
Please let me know if you had any other scenarios in mind

maytasm · 2022-07-27T22:05:02Z

@maytasm, thanks for the review.

Can you also add some Integration test (See integration test group "kinesis-index")?

Sure, I can add a simple test where the records are queried after ingesting from a stream with at least one empty shard. Please let me know if you had any other scenarios in mind

Sounds good to me! Thanks!

maytasm · 2022-07-29T21:01:48Z

@AmatyaAvadhanula
I have some questions regarding KinesisRecordSupplier#getSequenceNumber.

Are we still handling the case when shardIterator returned by kinesis.getShardIterator is null?
Is the timeoutMillis still needed? Is the timeoutMillis use for handling when shards are temporary closed due to SplitShard or MergeShards operation?

maytasm · 2022-07-29T22:35:34Z

...afka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java

@@ -157,6 +158,14 @@ public Long getEarliestSequenceNumber(StreamPartition<Integer> partition)
    return nextPos;
  }

+  @Override
+  public boolean isOffsetAvailable(StreamPartition<Integer> partition, OrderedSequenceNumber<Long> offset)


Does this needs a lock on recordSupplierLock?

Yes, thanks!

maytasm · 2022-07-29T22:50:17Z

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java

@@ -666,6 +670,30 @@ public String getEarliestSequenceNumber(StreamPartition<String> partition)
    return getSequenceNumber(partition, ShardIteratorType.TRIM_HORIZON);
  }

+  @Override
+  public boolean isOffsetAvailable(StreamPartition<String> partition, OrderedSequenceNumber<String> offset)


This is quite different from the current logic in KinesisRecordSupplier#getSequenceNumber.
such as handling when closed or the Retry for kinesis.getRecords

The previous exception handling has been added

AmatyaAvadhanula · 2022-08-01T06:19:25Z

Can you also add some Integration test (See integration test group "kinesis-index")?

Added a few tests where empty shards are present

Are we still handling the case when shardIterator returned by kinesis.getShardIterator is null?

Yes, an end of shard marker is still being returned. If this doesn't happen immediately, the task will handle it appropriately after assignment.

Is the timeoutMillis still needed? Is the timeoutMillis use for handling when shards are temporary closed due to SplitShard or MergeShards operation?

Could you please clarify what timeoutMillis is?
If you mean fetchSequenceNumberTimeout, it is the time that we poll kinesis for records before determining that it is empty. This was needed as we had to get at least one record before beginning ingestion and kinesis may not return records at TRIM_HORIZON or LATEST in the first few iterations depending on when data was added.
It isn't needed now as the new marker is returned and the task will handle it appropriately.
This prevents multiple retries to fetch records and makes it faster as well.

maytasm

Since this include a Design Review, we should get one more approval from a commiter

abhishekagarwal87 · 2022-08-04T04:23:46Z

docs/development/extensions-core/kinesis-ingestion.md

@@ -300,7 +300,6 @@ The `tuningConfig` is optional. If no `tuningConfig` is specified, default param
 |`recordBufferSize`|Integer|Size of the buffer (number of events) used between the Kinesis fetch threads and the main ingestion thread.|no (default == 10000)|
 |`recordBufferOfferTimeout`|Integer|Length of time in milliseconds to wait for space to become available in the buffer before timing out.| no (default == 5000)|
 |`recordBufferFullWait`|Integer|Length of time in milliseconds to wait for the buffer to drain before attempting to fetch records from Kinesis again.|no (default == 5000)|
-|`fetchSequenceNumberTimeout`|Integer|Length of time in milliseconds to wait for Kinesis to return the earliest or latest sequence number for a shard. Kinesis will not return the latest sequence number if no data is actively being written to that shard. In this case, this fetch call will repeatedly timeout and retry until fresh data is written to the stream.|no (default == 60000)|


what happens if someone has this parameter in their ingestion spec?

It is ignored. (Just like any parameter which isn't relevant in the payload, if at all)

abhishekagarwal87 · 2022-08-04T04:23:50Z

...indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskRunner.java

@@ -128,8 +128,7 @@ protected void possiblyResetDataSourceMetadata(
      for (final StreamPartition<String> streamPartition : assignment) {
        String sequence = currOffsets.get(streamPartition.getPartitionId());
        String earliestSequenceNumber = recordSupplier.getEarliestSequenceNumber(streamPartition);


can we pass this to the isOffsetAvailable method?

since its an expensive call.

It's no longer being used here. Thanks for catching this!

abhishekagarwal87 · 2022-08-04T04:30:54Z

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java

      );
  }

  @Override
  public int compareTo(OrderedSequenceNumber<String> o)
  {
    KinesisSequenceNumber num = (KinesisSequenceNumber) o;
+    if (isUnread() && num.isUnread()) {


shouldn't we fallback to comparing maxSequenceNumber in this case as well?

It's a boolean (isMaxSequenceNumber) which is used later. There is only one value to be compared per sequence number

AmatyaAvadhanula · 2022-08-04T11:33:17Z

@abhishekagarwal87 thanks for review!

Could you please verify the changes in the latest commit as well? It tries to fall back to earliest sequence number comparison in case the primary check for kinesis offset availability fails.

This reverts commit 857d0a9.

abhishekagarwal87 · 2022-08-05T17:09:27Z

Thank you @AmatyaAvadhanula. I have merged your change.

abhishekagarwal87 · 2022-08-09T11:00:49Z

@AmatyaAvadhanula
Now that kinesis ingestion works with empty shards, we need to remove this limitation from the docs.

Kinesis ingestion with empty shards

00f563f

AmatyaAvadhanula mentioned this pull request Jul 15, 2022

Fix and document skipIgnorableShards #12791

Closed

Fix tests

ad915f7

maytasm reviewed Jul 27, 2022

View reviewed changes

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java Show resolved Hide resolved

maytasm reviewed Jul 27, 2022

View reviewed changes

...-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java Show resolved Hide resolved

maytasm requested changes Jul 27, 2022

View reviewed changes

AmatyaAvadhanula added 2 commits July 27, 2022 13:20

Clean up fetchSequenceNumberTimeout

7befac5

Add docs for isMoreToReadBeforeReadingRecord

7d83583

AmatyaAvadhanula added 2 commits July 29, 2022 15:46

Fix tests and use a better sequence availability check

adb633c

Modify KinesisSupervisorTest after refactoring

3567466

maytasm added Design Review Area - Streaming Ingestion AWS Kinesis For changes in Kinesis ingestion labels Jul 29, 2022

maytasm reviewed Jul 29, 2022

View reviewed changes

AmatyaAvadhanula added 2 commits August 1, 2022 10:55

Lock record supplier usage in supervisor

604e64a

Kinesis integration tests for empty shard scenarios

d88e464

AmatyaAvadhanula added 2 commits August 1, 2022 12:46

wrap exception

fbc6064

Add check and fix offset exclusive clause

b041f43

maytasm approved these changes Aug 2, 2022

View reviewed changes

Add unit test

cde82e9

abhishekagarwal87 reviewed Aug 4, 2022

View reviewed changes

abhishekagarwal87 approved these changes Aug 4, 2022

View reviewed changes

Add secondary check for kinesis offset availability

857d0a9

AmatyaAvadhanula added 2 commits August 5, 2022 14:53

Revert "Add secondary check for kinesis offset availability"

c879a6e

This reverts commit 857d0a9.

Remove unnecessary call, add getRecords reference

982f0ab

abhishekagarwal87 added the Release Notes label Aug 5, 2022

abhishekagarwal87 merged commit d294404 into apache:master Aug 5, 2022

AmatyaAvadhanula mentioned this pull request Aug 10, 2022

Kinesis docs and logs improvements #12886

Merged

2 tasks

abhishekagarwal87 added this to the 24.0.0 milestone Aug 26, 2022

techdocsmith mentioned this pull request Aug 26, 2022

[Draft] 24.0 Release notes #12825

Closed

abhishekagarwal87 mentioned this pull request Sep 8, 2022

Test issue [Please ignore] #13055

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kinesis ingestion with empty shards #12792

Kinesis ingestion with empty shards #12792

AmatyaAvadhanula commented Jul 15, 2022

maytasm Jul 27, 2022

AmatyaAvadhanula Jul 27, 2022

AmatyaAvadhanula Aug 1, 2022

maytasm Jul 27, 2022

AmatyaAvadhanula Jul 27, 2022

AmatyaAvadhanula Aug 1, 2022

maytasm left a comment

AmatyaAvadhanula commented Jul 27, 2022 •

edited

maytasm commented Jul 27, 2022

maytasm commented Jul 29, 2022

maytasm Jul 29, 2022

AmatyaAvadhanula Aug 1, 2022

maytasm Jul 29, 2022

AmatyaAvadhanula Aug 1, 2022

AmatyaAvadhanula commented Aug 1, 2022

maytasm left a comment

abhishekagarwal87 Aug 4, 2022

AmatyaAvadhanula Aug 4, 2022

abhishekagarwal87 Aug 4, 2022

abhishekagarwal87 Aug 4, 2022

AmatyaAvadhanula Aug 4, 2022

abhishekagarwal87 Aug 4, 2022

AmatyaAvadhanula Aug 4, 2022 •

edited

AmatyaAvadhanula commented Aug 4, 2022

abhishekagarwal87 commented Aug 5, 2022

abhishekagarwal87 commented Aug 9, 2022

Kinesis ingestion with empty shards #12792

Kinesis ingestion with empty shards #12792

Conversation

AmatyaAvadhanula commented Jul 15, 2022

Description

Key changed/added classes in this PR

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

maytasm left a comment

Choose a reason for hiding this comment

AmatyaAvadhanula commented Jul 27, 2022 • edited

maytasm commented Jul 27, 2022

maytasm commented Jul 29, 2022

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

AmatyaAvadhanula commented Aug 1, 2022

maytasm 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

AmatyaAvadhanula Aug 4, 2022 • edited

Choose a reason for hiding this comment

AmatyaAvadhanula commented Aug 4, 2022

abhishekagarwal87 commented Aug 5, 2022

abhishekagarwal87 commented Aug 9, 2022

AmatyaAvadhanula commented Jul 27, 2022 •

edited

AmatyaAvadhanula Aug 4, 2022 •

edited