《Java 消息中间件·可靠消息与消费治理》3.1 事务生产者与 read_committed

The TransactionalId to use for transactional delivery. This enables reliability semantics which span
multiple producer sessions since it allows the client to guarantee that transactions using the same
TransactionalId have been completed prior to starting any new transactions. If no TransactionalId is
provided, then the producer is limited to idempotent delivery. If a TransactionalId is configured,
enable.idempotence is implied. By default the TransactionId is not configured, which means
transactions cannot be used. Note that, by default, transactions require a cluster of at least three
brokers which is the recommended setting for production; for development you can change this, by
adjusting broker setting transaction.state.log.replication.factor.

-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1

public void initTransactions() {
    throwIfNoTransactionManager();
    throwIfProducerClosed();
    long now = time.nanoseconds();
    TransactionalRequestResult result = transactionManager.initializeTransactions(false);
    sender.wakeup();
    result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, INIT_TXN_TIMEOUT_MSG);
    producerMetrics.recordInit(time.nanoseconds() - now);
    transactionManager.maybeUpdateTransactionV2Enabled(true);
}

FIND_COORDINATOR  clientId=producer-libraryd-apicheck-2  keyType=1 coordinatorKeys=[libraryd-apicheck-2]
INIT_PRODUCER_ID  clientId=producer-libraryd-apicheck-2  producerId=-1 producerEpoch=-1 transactionTimeoutMs=60000
PRODUCE           clientId=producer-libraryd-apicheck-2  acks=-1
END_TXN           clientId=producer-libraryd-apicheck-2  producerId=10 producerEpoch=0 committed=true

Feature: transaction.version  SupportedMinVersion: 0  SupportedMaxVersion: 2  FinalizedVersionLevel: 2

public synchronized void beginTransaction() {
    ensureTransactional();
    throwIfPendingState(TransactionOperation.BEGIN_TRANSACTION);
    maybeFailWithError();
    transitionTo(State.IN_TRANSACTION);
}

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19192");
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, txnId);
p.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000");
p.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
    producer.initTransactions();
    producer.beginTransaction();
    for (int i = 0; i < count; i++) {
        RecordMetadata md = producer.send(
                new ProducerRecord<>(topic, "book-" + i, prefix + "-" + i)).get();
        System.out.println("sent value=" + prefix + "-" + i + " offset=" + md.offset());
    }
    if ("commit".equals(action)) {
        producer.commitTransaction();
    } else {
        producer.abortTransaction();
    }
}

initTransactions ok txnId=libraryd-txn-1
beginTransaction ok
sent value=committed-0 partition=0 offset=0
sent value=committed-1 partition=0 offset=1
commitTransaction ok

initTransactions ok txnId=libraryd-txn-1
beginTransaction ok
sent value=aborted-0 partition=0 offset=3
abortTransaction ok

isolation=read_committed topic=library.loan.created.txn beginning=0 endOffsets=5
  off=0 key=book-0 value=committed-0
  off=1 key=book-1 value=committed-1
read total=2 position=5 endOffsets=5

isolation=read_uncommitted topic=library.loan.created.txn beginning=0 endOffsets=5
  off=0 key=book-0 value=committed-0
  off=1 key=book-1 value=committed-1
  off=3 key=book-0 value=aborted-0
read total=3 position=5 endOffsets=5

baseOffset: 0 count: 1 baseSequence: 0 lastSequence: 0 producerId: 0 producerEpoch: 0 isTransactional: true isControl: false
| offset: 0 key: book-0 payload: committed-0
baseOffset: 1 count: 1 baseSequence: 1 lastSequence: 1 producerId: 0 producerEpoch: 0 isTransactional: true isControl: false
| offset: 1 key: book-1 payload: committed-1
baseOffset: 2 count: 1 baseSequence: -1 lastSequence: -1 producerId: 0 producerEpoch: 1 isTransactional: true isControl: true
| offset: 2 endTxnMarker: COMMIT coordinatorEpoch: 0
baseOffset: 3 count: 1 baseSequence: 0 lastSequence: 0 producerId: 0 producerEpoch: 2 isTransactional: true isControl: false
| offset: 3 key: book-0 payload: aborted-0
baseOffset: 4 count: 1 baseSequence: -1 lastSequence: -1 producerId: 0 producerEpoch: 3 isTransactional: true isControl: true
| offset: 4 endTxnMarker: ABORT coordinatorEpoch: 0

# 事务打开期间，read_committed
isolation=read_committed topic=library.loan.open beginning=0 endOffsets=0
read total=0 position=0 endOffsets=0

# 事务打开期间，read_uncommitted
isolation=read_uncommitted topic=library.loan.open beginning=0 endOffsets=1
  off=0 key=book-0 value=open-0
read total=1 position=1 endOffsets=1

# 提交之后，read_committed
isolation=read_committed topic=library.loan.open beginning=0 endOffsets=2
  off=0 key=book-0 value=open-0
read total=1 position=2 endOffsets=2

If set to read_committed, consumer.poll() will only return transactional messages which have been
committed. ... Messages will always be returned in offset order. Hence, in read_committed mode,
consumer.poll() will only return messages up to the last stable offset (LSO), which is the one less
than the offset of the first open transaction. In particular any messages appearing after messages
belonging to ongoing transactions will be withheld until the relevant transaction has been completed.
As a result, read_committed consumers will not be able to read up to the high watermark when there
are in flight transactions.

bootstrap.servers=127.0.0.1:19192
group.id=libraryd-reader
enable.auto.commit=false
isolation.level=read_committed
auto.offset.reset=earliest