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

[sandeep-mst]

Fixes apache#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:

[sandeep-mst]

Closing this in favour of PR in connectors repository.
apache/pulsar-connectors#11