Savonitar opened a new pull request, #133: URL: https://github.com/apache/flink-connector-pulsar/pull/133
## Purpose of the change Follow-up to #130 and the [review comment about negative message IDs](https://github.com/apache/flink-connector-pulsar/pull/130#discussion_r4346450312). Pulsar can successfully acknowledge a deduplicated transactional send with ledger and entry IDs of `-1` without appending a message. The writer currently records this sentinel when a checkpoint contains only duplicate sends, leaving an invalid published position in its committable. In Pulsar 4.0.14, the [transactional duplicate-send branch](https://github.com/apache/pulsar/blob/v4.0.14/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L5005-L5008) explicitly calls `publishContext.completed(null, -1, -1)`. The [client acknowledgement handler](https://github.com/apache/pulsar/blob/v4.0.14/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1497-L1515) retains these IDs and completes the send successfully. Pulsar's [deduplication test](https://github.com/apache/pulsar/blob/v4.0.14/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java#L365-L373) also asserts that a successful duplicate send returns `-1` for both ledger and entry IDs. Skip acknowledgements with a negative ledger or entry ID when tracking published positions, while still emitting the transaction committable. This fixes checkpoint metadata; the current committer continues to complete transactions using their transaction IDs. An empty position map means no position is available. It can represent restored V1 state or a checkpoint containing only deduplicated sends; it does not determine the transaction outcome. ## Brief change log - Filter negative ledger and entry IDs before updating the per-topic position map. - Add broker-backed coverage with batching enabled and disabled for duplicate-only and mixed checkpoints. - Verify that every duplicate send leaves the ledger position unchanged after flushing and before committing, and that mixed sends retain the actual persisted message ID. ## Verifying this change ``` mvn -pl flink-connector-pulsar -Dtest=PulsarWriterTest,ProducerRegisterTest,PulsarCommittableSerializerTest -Dflink.forkCountUnitTest=1 test ``` - All 23 targeted tests passed against Pulsar 4.0.14. - Removing only the production filter makes both regression cases fail because the duplicate-only committable contains `ledgerId=-1, entryId=-1`; restoring it makes them pass. - Checkstyle passed. Spotless checks on the changed files passed using Java 11. - Full repository verification and the E2E suite were not run. ## Significant changes - [ ] Dependencies have been added or upgraded - [ ] Public API has been changed - [ ] Serializers have been changed - [ ] New feature has been introduced --- ##### Was generative AI tooling used to co-author this PR? - [x] Yes Generated-by: OpenAI Codex (GPT-6) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
