Motivation

The current Kinesis source connector violates the at-least-once delivery guarantee, which can lead to data loss during failures or restarts. The root cause is that the Kinesis checkpointing is based on a time interval, completely decoupled from Pulsar's acknowledgment mechanism.

This creates a race condition that leads to data loss, as illustrated below:

• Time 1: KinesisRecordProcessor fetches record 0 from a Kinesis shard and puts it into the internal queue.
• Time 2: The KinesisSource thread reads record 0 from the queue and attempts to send it to a Pulsar topic, but the send operation is delayed or failing.
• Time 3: The KinesisRecordProcessor's timer triggers. It checkpoints its progress, marking that it has successfully processed up to record 0, even though this record has not yet been acknowledged by Pulsar.
• Time 4: The connector crashes before record 0 can be successfully sent and acked. Upon restart, the KCL worker reads the last checkpoint and starts consuming from the position after record 0, causing record 0 to be permanently lost.

This PR fixes this critical issue by ensuring that checkpoints are only committed after a record has been successfully acknowledged by the Pulsar IO runtime.

Modifications

1. Refactored Checkpointing Logic: The checkpointing mechanism in KinesisRecordProcessor was changed from being time-based to being driven by Pulsar acks. It now checkpoints a specific sequenceNumber provided by an acknowledged record.

2. Implemented ack() and fail(): In KinesisRecord.java, the ack() method was implemented to call back to the KinesisRecordProcessor and update the latest sequence number that is safe to checkpoint. The fail() method now calls sourceContext.fatal() to ensure the connector restarts upon failure.

3. Passed SourceContext: The SourceContext is now passed down to the KinesisRecordProcessor to allow it to trigger a fatal connector restart upon unrecoverable errors (e.g., failed checkpoint attempts).

Verifying this change

• Added Unit Tests: Created KinesisRecordProcessorTest.java to verify the new ack-driven checkpointing logic.

Integration test

1. Create a pulsar topic and a subscription.
2. Create a Kinesis source connector.
3. Send 100 messages to Kinesis with 1s internal.

    public static void main(String[] args) throws Exception {

        AWSCredentialsProvider credentialsProvider =
                new AWSStaticCredentialsProvider(new BasicAWSCredentials("", ""));

        KinesisProducerConfiguration kinesisConfig = new KinesisProducerConfiguration();
        kinesisConfig.setRegion("ap-northeast-1");
        kinesisConfig.setCredentialsProvider(credentialsProvider);
        KinesisProducer kinesis = new KinesisProducer(kinesisConfig);
        // Put some records 
        for (int i = 0; i < 100; ++i) {
            ByteBuffer data = ByteBuffer.wrap("test-kinesis-data".getBytes("UTF-8"));
            // doesn't block       
            kinesis.addUserRecord("connector-source", "myPartitionKey", data);
            kinesis.flush();
            Thread.sleep(1000);
        }
        kinesis.flush();
        Thread.sleep(10000);
        kinesis.destroy();
    }

4. Topic unloads and connector restarts will occur at random intervals during this time.

Result:

Before this PR: will receive fewer than 100 messages

Receive message test-kinesis-data with count 96
Receive message test-kinesis-data with count 97
Received 97 messages

After this PR: *Verify that the number of received messages is greater than or equal to 100, which fulfills the at-least-once delivery guarantee.

Receive message test-kinesis-data with count 101
Receive message test-kinesis-data with count 102
Received 102 messages

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: