《Java 消息中间件·可靠消息与消费治理》3.2 事务超时、fencing 与 abort

The maximum amount of time in milliseconds that a transaction will remain open before the coordinator
proactively aborts it. The start of the transaction is set at the time that the first partition is
added to it. If this value is larger than the transaction.max.timeout.ms setting in the broker, the
request will fail with a InvalidTxnTimeoutException error.

transaction.abort.timed.out.transaction.cleanup.interval.ms=10000
transaction.max.timeout.ms=900000
transaction.state.log.num.partitions=50

p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, txnId);
p.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, Integer.toString(5000));
p.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "20000");

producer.initTransactions();
producer.beginTransaction();
for (int i = 0; i < 2; i++) {
    RecordMetadata md = producer.send(
            new ProducerRecord<>(topic, "book-" + i, "timeout-" + i)).get();
    System.out.println("sent value=timeout-" + i + " offset=" + md.offset());
}
Thread.sleep(20000);
try {
    producer.commitTransaction();
    System.out.println("commitTransaction UNEXPECTED ok");
} catch (Exception e) {
    System.out.println("commit FAILED: " + e.getClass().getName() + ": " + e.getMessage());
}

initTransactions ok transaction.timeout.ms=5000
sent value=timeout-0 offset=0 at 1793250941798
sent value=timeout-1 offset=1 at 1793250941809
TXN_OPEN sleeping 20000 ms
commit FAILED after 15 ms: org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted a transactional operation in an invalid state.
begin after failure FAILED: org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
producer closed

baseOffset: 0 producerId: 2 producerEpoch: 0 isTransactional: true isControl: false
| offset: 0 payload: timeout-0
baseOffset: 1 producerId: 2 producerEpoch: 0 isTransactional: true isControl: false
| offset: 1 payload: timeout-1
baseOffset: 2 producerId: 2 producerEpoch: 1 isTransactional: true isControl: true
| offset: 2 endTxnMarker: ABORT coordinatorEpoch: 0

$ kafka-transactions.sh --bootstrap-server 127.0.0.1:9092 list
TransactionalId      Coordinator  ProducerId  TransactionState
libraryd-timeout-1   1            2           CompleteAbort

isolation=read_committed topic=library.loan.timeout beginning=0 endOffsets=3
read total=0 position=3 endOffsets=3

isolation=read_uncommitted topic=library.loan.timeout beginning=0 endOffsets=3
  off=0 key=book-0 value=timeout-0
  off=1 key=book-1 value=timeout-1
read total=2 position=3 endOffsets=3

TRACE [KafkaApi-1] Handling request: apiKey=END_TXN ... transactionalId='libraryd-timeout-2'
  producerId=13 producerEpoch=0 committed=true
DEBUG [TransactionCoordinator id=1] Aborting append of COMMIT to transaction log with coordinator
  and returning INVALID_TXN_STATE error to client for libraryd-timeout-2's EndTransaction request
TRACE [KafkaApi-1] Completed libraryd-timeout-2's EndTxnRequest with committed: true,
  errors: INVALID_TXN_STATE from client producer-libraryd-timeout-2.

} else if (error == Errors.INVALID_PRODUCER_EPOCH || error == Errors.PRODUCER_FENCED) {
    // We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,
    // just treat it the same as PRODUCE_FENCED.
    fatalError(Errors.PRODUCER_FENCED.exception());
} else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
        error == Errors.INVALID_TXN_STATE || error == Errors.INVALID_PRODUCER_ID_MAPPING) {
    fatalError(error.exception());
}

isAcked = true;
if (error != null) {
    throw error;
}

if (lastError instanceof ProducerFencedException) {
    throw new ProducerFencedException("Producer with transactionalId '" + transactionalId
            + "' and " + producerIdAndEpoch + " has been fenced by another producer "
            + "with the same transactionalId");
}

KafkaProducer<String, String> a = producer(txnId);
KafkaProducer<String, String> b = producer(txnId);

a.initTransactions();
a.beginTransaction();
a.send(new ProducerRecord<>(topic, "book-1", "from-A")).get();

b.initTransactions();          // 同一个 transactional.id，推高 epoch
b.beginTransaction();
b.send(new ProducerRecord<>(topic, "book-1", "from-B")).get();
b.commitTransaction();

try {
    a.commitTransaction();     // 应被 fence
} catch (Exception e) {
    System.out.println("A commit FAILED: " + e.getClass().getName() + ": " + e.getMessage());
}

A initTransactions ok
A sent from-A offset=0 (txn still open)
B initTransactions ok (same transactional.id)
B sent from-B offset=2
B commitTransaction ok
A commit FAILED: org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one.
A send after fence FAILED: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.ProducerFencedException: Producer with transactionalId 'libraryd-fence-1' and (producerId=3, epoch=0) has been fenced by another producer with the same transactionalId

offset 0  producerId=3 producerEpoch=0 isControl=false  payload: from-A
offset 1  producerId=3 producerEpoch=1 isControl=true   endTxnMarker: ABORT
offset 2  producerId=3 producerEpoch=2 isControl=false  payload: from-B
offset 3  producerId=3 producerEpoch=3 isControl=true   endTxnMarker: COMMIT

isolation=read_committed topic=library.loan.fence beginning=0 endOffsets=4
  off=2 key=book-1 value=from-B
read total=1 position=4 endOffsets=4

isolation=read_uncommitted topic=library.loan.fence beginning=0 endOffsets=4
  off=0 key=book-1 value=from-A
  off=2 key=book-1 value=from-B
read total=2 position=4 endOffsets=4