[fix][io] Fix acknowledgments not being sent when collapsePartitionedTopics=true in kafka-connect-adapter

[sandeep-mst]

Fixes 25450

Motivation

Fix acknowledgments not being sent when the collapsePartitionedTopics=true on kafka-connect-adapter. The topic name stored in sinkrecord and currentOffsets is different from the topic name stored in the pulsar record.

The topic name used in SinkRecord / currentOffsets is collapsed (base topic name) and the topic name in the original Pulsar Record remains non-collapsed (includes -partition-X)

As a result, during ackUntil()

• Topic lookup in committedOffsets fails
• lastCommittedOffset will always be null because of topic name mismatch.
• Records are never acknowledged

Modifications

Updated ackUntil() to derive topic and partition from the source Record using the same logic as toSinkRecord() (i.e., respecting collapsePartitionedTopics)
Added a debug log and ackRequestedCount for better logging clarity.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• org.apache.pulsar.io.kafka.connect.KafkaConnectSinkTest#testAckUntilWithCollapsePartitionedTopics
• org.apache.pulsar.io.kafka.connect.KafkaConnectSinkTest#testAckUntilWithoutCollapsePartitionedTopics

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository:
cognitree#1

[lhotari]

@sandeep-mst Please rebase / merge latest master. I didn't have permission to do that for your PR branch.

❯ gh pr update-branch 11
GraphQL: user doesn't have permission to update head repository (updatePullRequestBranch)

[sandeep-mst]

@lhotari done.

[Copilot]

Pull request overview

Fixes missing acknowledgments in the Kafka Connect sink when collapsePartitionedTopics=true by ensuring ackUntil() derives the Kafka topic/partition the same way as toSinkRecord(), avoiding topic-name mismatches for partitioned topics.

Changes:

• Update ackUntil() to compute topic + partition using the same collapse logic as toSinkRecord().
• Factor collapse detection into a shared helper and improve debug logging around acknowledgments.
• Add unit tests covering ackUntil() behavior with collapsePartitionedTopics enabled/disabled.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java	Align ackUntil() topic/partition derivation with toSinkRecord() via a shared helper; add debug logging.
kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java	Add focused tests validating ackUntil() ack behavior for collapsed vs non-collapsed partitioned topics.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[lhotari]

@sandeep-mst Please check the review comments

[sandeep-mst]

@lhotari All the review comments have been resolved. Please review when possible.

Thanks and Regards

[lhotari]

LGTM, good work @sandeep-mst

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.