poorbarcode commented on code in PR #132:
URL:
https://github.com/apache/flink-connector-pulsar/pull/132#discussion_r4451315446
##########
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:
Fixed in fef8303. V2 committables with message IDs now use a separate error
handler. `InvalidTxnStatusException` and `TransactionNotFoundException` trigger
bounded retries via `retryLater()`, and each retry starts by querying message
state again. Even at the retry limit, confirmed committed state succeeds; an
unconfirmed outcome fails with the original commit exception as its cause. V1
retains its legacy handling.
Confirmed aborts, expired message state, and exhausted admin-query retries
now throw `FlinkRuntimeException`. In Flink 2.0.2,
`signalFailedWithKnownReason()` marks a committable as terminally failed
without throwing, allowing processing to continue. Throwing propagates the
failure to the task instead of discarding the affected output.
Added tests for abort races, commits that completed despite a returned
error, first/subsequent retries, zero/exhausted retry budgets, and
unavailable/expired verification state. All 67 serializer/committer tests pass,
as does the full-repository Spotless check.
--
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]
