apache / apache/beam

[Task]: Adjust KinesisIO ShardCheckpoint semantics such that is more consistent

Open
#26,073 0 comments 0 reactions 0 assignees View on GitHub
aws java kinesis P3 task
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
2d 2h
Merged PRs (30d)
205

Description

### What needs to happen?

https://github.com/apache/beam/blob/7ee74d2bf7338e82d35e4429e6d21decc1097621/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisReader.java always creates checkpoints with AFTER_SEQUENCE_NUMBER as soon as a record for a shard is ack-ed.

When the consumer starts from a checkpoint, such checkpoint is always "converted" into AT_SEQUENCE_NUMBER for shard reading, because subSequenceNumber is never set to `null` in records acknowledgements: https://github.com/apache/beam/blob/7ee74d2bf7338e82d35e4429e6d21decc1097621/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardCheckpoint.java#L155

This commit demonstrates it: https://github.com/psolomin/beam/pull/114/commits/93f29a7447e76a1d366db7f54170e5cceadb5797

It would be more consistent to persist AT_SEQUENCE_NUMBER in checkpoints, and have the consumers starting points to be used directly from checkpoints.

This is how the proposed change might look like: https://github.com/psolomin/beam/pull/114/commits/18b4d537fe494df369e95c2585537e28cc90082c

### Issue Priority

Priority: 3 (nice-to-have improvement)

### Issue Components

- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Contributor guide

Open the contributing guide

Research direction

Start with sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisReader.java and ShardCheckpoint.java around line 155. Read the linked commits to understand the proposed checkpoint and consumer-starting-point behavior. Done means checkpoint persistence and shard reading use consistent sequence-number semantics, with the relevant Kinesis IO tests updated or verified.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering
Issue type
Refactor
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.