DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
- Producer.sendShareAcknowledgementsToTransaction(Map<TopicIdPartition, List<AcknowledgementBatch>>, ShareGroupMetadata).
- ShareGroupMetadata class (groupId, memberId, memberEpoch, optional groupInstanceId).
- ShareConsumer.shareGroupMetadata() to obtain the above.
| Code Block |
|---|
Producer (clients module) public interface Producer<K, V> { /** * Sends a list of share-group acknowledgements to the consumer-coordinator and marks * them for atomic commit alongside the records produced in this transaction. * * The acknowledgements are staged on the broker in a TX_PENDING state until the * transaction is committed or aborted. On commit, ACCEPT records transition to * ACKNOWLEDGED and REJECT records transition to ARCHIVING/ARCHIVED. On abort, * all staged records revert to ACQUIRED and remain owned by the original consumer * until either the consumer re-acknowledges them or the acquisition lock expires. * * @param acknowledgements Per-partition list of acknowledgement batches. Only * AcknowledgeType.ACCEPT (1) and AcknowledgeType.REJECT (3) * are valid inside a transaction. RELEASE (2), RENEW (4), * and GAP (0) are rejected with InvalidRecordStateException. * @param groupMetadata Snapshot of the share consumer's group identity, obtained * from ShareConsumer.shareGroupMetadata(). * * @throws IllegalStateException if no transaction is in progress, or if the producer * is not transactional. * @throws ProducerFencedException if another producer with the same transactionalId * has fenced this one. * @throws UnsupportedVersionException if the cluster does not advertise apiKey 93 * (TxnShareAcknowledge) via ApiVersions. * @throws GroupAuthorizationException if the configured principal cannot Write to * the share group. * @throws InvalidProducerEpochException if the producer's epoch is stale. * @throws KafkaException for other non-fatal errors that may be retried by aborting * the transaction and retrying the entire read-process-write * loop. * * Threading: Returns immediately after enqueuing the request for the producer's * background Sender thread. The broker has NOT processed it yet. * Any error is reported when commitTransaction() or abortTransaction() is later called. */ void sendShareAcknowledgementsToTransaction( Map<TopicIdPartition, List<AcknowledgementBatch>> acknowledgements, ShareGroupMetadata groupMetadata ) throws ProducerFencedException; } |
| Code Block |
|---|
ShareConsumer (clients module)
public interface ShareConsumer<K, V> {
/**
* Returns an immutable snapshot of this consumer's share-group identity for use
* with Producer.sendShareAcknowledgementsToTransaction.
*
* The snapshot captures groupId, memberId, and memberEpoch atomically; if a
* rebalance changes the memberEpoch between the snapshot and the producer call,
* the broker will reject the staging request with STALE_MEMBER_EPOCH and the
* user must abort the transaction and retry the read-process-write loop.
*
* @throws UnsupportedVersionException if the cluster does not support KIP-1289.
* @throws TimeoutException if the snapshot cannot be obtained within the
* configured default.api.timeout.ms.
*
* Threading: thread-safe; safe to call concurrently with poll() and acknowledge().
*/
ShareGroupMetadata shareGroupMetadata();
} |
Wire protocol additions:
...