Copilot commented on code in PR #132:
URL:
https://github.com/apache/flink-connector-pulsar/pull/132#discussion_r4451049208
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -112,7 +122,7 @@ public PulsarCommittable deserialize(int version, byte[]
serialized) throws IOEx
if (in.available() != 0) {
throw new IOException("Trailing bytes in Pulsar committable
state");
}
- return new PulsarCommittable(txnID, messages);
+ return new PulsarCommittable(txnID, messages, version);
Review Comment:
Reserializing a restored V1 committable writes V2 state without preserving
its legacy version, so another restore produces a V2 committable with no
message IDs. The committer then treats it as an empty transaction and
acknowledges commit failures as success, even though the legacy transaction may
contain output. Preserve legacy provenance across snapshots and add a V1
restore → reserialize → restore regression test that checks failed-commit
handling.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java:
##########
@@ -70,76 +80,177 @@ public PulsarCommitter(
@SuppressWarnings("java:S3776")
public void commit(Collection<CommitRequest<PulsarCommittable>> requests)
throws PulsarClientException {
- TransactionCoordinatorClient client = transactionCoordinatorClient();
+ if (requests.isEmpty()) {
+ return;
+ }
for (CommitRequest<PulsarCommittable> request : requests) {
- PulsarCommittable committable = request.getCommittable();
- TxnID txnID = committable.getTxnID();
- LOG.debug("Start committing the Pulsar transaction {}", txnID);
+ // Checkpoint state written by older connector versions does not
contain message IDs,
+ // so preserve its direct-commit behavior. Current state carries
message IDs and must
+ // always use them for recovery checks, including when several
checkpoints are
+ // recovered together.
+ if (request.getCommittable().getVersion() == 1) {
+ commitLegacyTransaction(request);
+ } else {
+ commitCurrentTransaction(request);
+ }
+ }
+ }
+
+ private void commitCurrentTransaction(CommitRequest<PulsarCommittable>
request)
+ throws PulsarClientException {
+ PulsarCommittable committable = request.getCommittable();
+ TxnID txnID = committable.getTxnID();
+ Map<String, MessageIdPojo> latestMessages =
committable.getLatestPublishedMessages();
+ TransactionCoordinatorClient client = transactionCoordinatorClient();
+
+ // A transaction without output has no observable message state. Its
outcome is irrelevant
+ // to the result, so a failure here must not make checkpoint recovery
fail.
+ if (latestMessages.isEmpty()) {
try {
client.commit(txnID);
- } catch (CoordinatorNotFoundException e) {
- LOG.error(
- "We couldn't find the Transaction Coordinator from
Pulsar broker {}. "
- + "Check your broker configuration.",
- committable,
- e);
- request.signalFailedWithKnownReason(e);
- } catch (InvalidTxnStatusException e) {
- LOG.error(
- "Unable to commit transaction ({}) because it's in an
invalid state. "
- + "Most likely the transaction has been
aborted for some reason. "
- + "Please check the Pulsar broker logs for
more details.",
- committable,
- e);
+ } catch (Exception e) {
+ LOG.warn("Failed to commit empty transaction {}.", txnID, e);
request.signalAlreadyCommitted();
- } catch (TransactionNotFoundException e) {
- if (request.getNumberOfRetries() == 0) {
- LOG.error(
- "Unable to commit transaction ({}) because it's
not found on Pulsar broker. "
- + "Most likely the checkpoint interval
exceed the transaction timeout.",
- committable,
- e);
- request.signalFailedWithKnownReason(e);
- } else {
- LOG.warn(
- "We can't find the transaction {} after {} retry
committing. "
- + "This may mean that the transaction have
been committed in previous but failed with timeout. "
- + "So we just mark it as committed.",
- txnID,
- request.getNumberOfRetries());
- request.signalAlreadyCommitted();
+ }
+ return;
+ }
+
+ int failedQueries = 0;
+ for (Map.Entry<String, MessageIdPojo> entry :
latestMessages.entrySet()) {
+ String topic = entry.getKey();
+ MessageIdPojo messageId = entry.getValue();
+ TopicName topicName = TopicName.get(topic);
+ String messageTopic =
+ messageId.getPartitionIndex() == -1 ||
topicName.isPartitioned()
+ ? topic
+ :
topicName.getPartition(messageId.getPartitionIndex()).toString();
+
+ final List<Message<byte[]>> messages;
+ try {
+ messages =
+ pulsarAdmin
+ .topics()
+ .getMessagesById(
+ messageTopic,
+ messageId.getLedgerId(),
+ messageId.getEntryId());
+ } catch (PulsarAdminException e) {
+ failedQueries++;
+ LOG.warn("Failed to query message {} on topic {}.", messageId,
messageTopic, e);
+ continue;
+ }
+
+ if (messages == null || messages.isEmpty()) {
+ continue;
+ }
+
+ Map<String, String> properties = messages.get(0).getProperties();
+ if ("true".equalsIgnoreCase(properties.get(TXN_ABORTED))) {
+ String message =
+ String.format(
+ "Transaction %s was aborted; messages: %s",
txnID, latestMessages);
+ LOG.warn(message);
+ request.signalFailedWithKnownReason(new
FlinkRuntimeException(message));
+ return;
+ }
+
+ if ("true".equalsIgnoreCase(properties.get(TXN_UNCOMMITTED))) {
+ try {
+ client.commit(txnID);
+ } catch (Exception e) {
+ handleError(request, txnID, committable, e);
}
Review Comment:
A transaction can time out and start aborting after the admin lookup reports
it as uncommitted but before `client.commit()`. The shared handler treats
`InvalidTxnStatusException`, or `TransactionNotFoundException` after a retry,
as success and can silently acknowledge aborted output. For V2 transactions,
retry message-state verification on these errors and fail conservatively when
retries are exhausted instead of using the legacy success heuristics.
--
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]
