[FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty #15
Conversation
…ucer This new flag tracks whether or not data was actually written to the current transaction
…as written to the transaction
…nsaction Tests that if the KafkaWriter is asked to prepareCommit but its current transaction is empty, it should: - NOT emit a KafkaCommittable - Immediately commit the empty transaction, and - Recycle the producer
|
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html) |
|
cc @RamanVerma @Gerrrr for review |
| snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch()); | ||
|
|
||
| try { | ||
| resumedProducer.commitTransaction(); |
There was a problem hiding this comment.
You can use assertThatThrownBy here
| @Override | ||
| public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) { | ||
| if (inTransaction) { | ||
| hasRecordsInTransaction = true; |
There was a problem hiding this comment.
Should this boolean be set in the callback, in the successful send scenario?
There was a problem hiding this comment.
hmm, that's a good point. I think the question to ask is: is it incorrect to set this flag (to allow a KafkaCommittable to be generated for the txn at pre-commit time) preemptively, instead of only setting it when data has actually been successfully written?
I think the answer is that it is not incorrect, so it is ok to leave this as is. Reasoning is as follows:
-
At pre-commit time and performing flush, if some data failed to be flushed, the pre-commit will fail so a
KafkaCommittablewill not be checkpointed for the txn anyways. In this scenario, thehasRecordsInTransactionflag is irrelevant no matter its value. -
If all records are correctly flushed, then good; a
KafkaCommittableshould be generated for the txn. We're good here because we've alraedy preemptively set thehasRecordsInTransactionflag.
There was a problem hiding this comment.
Speaking of which, it might give a more complete picture of this interaction after I rebase this PR branch on top of the latest changes (to include the fix that adds checkAsyncExceptions).
| resumedProducer.resumeTransaction( | ||
| snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch()); | ||
|
|
||
| assertThatThrownBy(resumedProducer::commitTransaction); |
There was a problem hiding this comment.
nit: Can we check the exception type also using
isInstanceOf(<exception type>.class)
|
thanks for the review @RamanVerma, merging this now! |
…rnalProducer This closes #15.
…rnalProducer This closes #15.
…rnalProducer This closes apache#15.
This PR fixes FLINK-31363 by changing how we handle empty transactions on
KafkaWriter#prepareCommit().Previously, regardless of whether the current transaction is empty or non-empty, we always emit a
KafkaCommittablefor it to be checkpointed by theCommitterOperator. The issue: on restore, when we resume the transaction and commit it, we recreate aFlinkKafkaInternalProducerthat always has the internaltransactionStartedflag set totrue, which means that anEndTxnRequestwill be sent to the brokers for committing the transaction. This results in anInvalidTxnStateerror since on the broker side the transaction hasn't actually been started yet (transactions are lazily started on brokers on the first record sent).I've considered two possible ways to address this:
transactionStartedflag in aKafkaCommittablealongside other txn metadata. Then, on restore, on the recreated producer, we set the internaltransactionStartedaccordingly to what the checkpoint says.KafkaCommitableif the transaction is empty. In this case, anyKafkaCommittablerestored from a checkpoint always has some data in them, and therefore it is correct to always set the internaltransactionStartedflag totrueon the recreated producer.This PR chooses to go with approach 2.
On
prepareCommit, if the current ongoing transaction is empty, then:KafkaCommittable(which will be checkpointed by the downstreamCommitterOperator)EndTxnRequestat all)