slachiewicz commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r4054032123
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/ProducerPoolImpl.java:
##########
@@ -254,7 +255,39 @@ public FlinkKafkaInternalProducer<byte[], byte[]>
getTransactionalProducer(
@Override
public Collection<CheckpointTransaction> getOngoingTransactions() {
- return new ArrayList<>(transactionalIdsByCheckpoint.keySet());
+ List<CheckpointTransaction> ongoing = new
ArrayList<>(transactionalIdsByCheckpoint.size());
+ for (Map.Entry<CheckpointTransaction, String> entry :
+ transactionalIdsByCheckpoint.entrySet()) {
+ CheckpointTransaction transaction = entry.getKey();
+ ProducerEntry producerEntry =
producerByTransactionalId.get(entry.getValue());
+ FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ producerEntry == null ? null : producerEntry.getProducer();
+ if (producer == null) {
+ // restored from state; keep what the state knows
+ ongoing.add(transaction);
+ } else {
+ ongoing.add(
+ new CheckpointTransaction(
+ transaction.getTransactionalId(),
+ transaction.getCheckpointId(),
+ producer.getProducerId(),
+ producer.getEpoch()));
+ }
+ }
+ return ongoing;
+ }
+
+ @Override
+ public void abandonTransaction(String transactionalId) {
+ ProducerEntry producerEntry =
producerByTransactionalId.get(transactionalId);
+ checkState(
+ producerEntry != null && producerEntry.getProducer() == null,
+ "Transaction %s is not a restored transaction without a
producer: %s",
+ transactionalId,
+ producerEntry);
+ producerByTransactionalId.remove(transactionalId);
Review Comment:
Yes, and it was more than a race: with `KafkaSinkRecoveryITCase` from #322
on top of this branch, POOLING took the id for the first post-restore
transaction and the fenced report then closed that producer, losing record 1.
`abandonTransaction` is now `abortRestoredTransaction`: it fences the
broker-side transaction through a producer that is never registered under the
id and keeps the restored entry, so the id stays in `getOngoingTransactions()`
until `recycleByTransactionId` releases it (the fenced report, or the sweep
once a later checkpoint finishes).
`ProducerPoolImplITCase#testAbortRestoredTransaction` and
`#testAbortRestoredTransactionReleasedBySweep` cover both release paths; the
recovery ITCase is green on 5 runs.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyImpl.java:
##########
@@ -143,6 +143,18 @@ public void abortTransactions(Context context) {
TransactionAborter transactionAborter =
context.getTransactionAborter();
for (String name : openTransactionsForSubtask) {
if (context.getPrecommittedTransactionalIds().contains(name)) {
+ if (context.isPrecommittedTransactionSuperseded(name)) {
+ // The broker holds a later transaction under this id
than the one the
+ // committer is about to commit. That commit will be
fenced, and nobody
+ // owns the open transaction; abort it so that it does
not pin the last
+ // stable offset until the transaction timeout.
+ LOG.warn(
Review Comment:
One line now: the WARN in the strategy is gone and the context logs a single
WARN with the recorded and reported producer id, epoch and state, where the
values are known.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
+ CheckpointTransaction transaction =
precommittedTransactions.get(transactionalId);
+ if (transaction == null || !transaction.hasKnownEpoch()) {
+ // state written before v3 did not record the epoch; keep the old
behavior
+ return false;
+ }
+ TransactionDescription description =
describePrecommitted().get(transactionalId);
+ if (description == null) {
+ // the broker does not know the id any more; nothing to abort
+ return false;
+ }
+ boolean superseded =
+ description.producerId() != transaction.getProducerId()
+ || description.producerEpoch() >
transaction.getEpoch();
+ if (superseded) {
+ LOG.info(
+ "Recovered transaction {} was opened with producer id {}
and epoch {}, but the broker now holds producer id {} and epoch {} in state {}",
Review Comment:
You are right, they are two separate observations. The message now says "the
broker reported ... in state {}" and nothing in the diff calls the transaction
open any more; the Context javadoc and the strategy comment were changed the
same way.
--
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]
