Preferred method of setting the next offset manually for Consumers with EOS semantic

[Moritz Ulmer (moritz89)]

Is there a difference when committing a consumer's offset between the two approaches? I Used to use the TopicPartition() approach but recently found the consumer.position() approach in the docs:

next_offset = consumer.position(consumer.assignment())

OR

consumed_message: KafkaMessage
next_offset = TopicPartition(
    consumed_message.topic(),
    consumed_message.partition(),
    consumed_message.offset() + 1,
)

This is used when sending offsets for EOS consumed messages.

# Send messages resulting from the processed message
# At the end of consumer's callback for its while-true loop
next_offset = consumer.position(consumer.assignment())
try:
    send_eos_messages(out_messages, next_offset, consumer.consumer_group_metadata())
    _decay_send_error()
except KWPostAbortTransaction:
    # When a transaction has been aborted, rewind the consumer's offset. By
    # decrementing the offset, the current message will be processed again.
    next_offset[0].offset -= 1
    consumer.commit(offsets=next_offset)
    _increment_send_error()

Context for the above send messages code snippet

import os
from dataclasses import dataclass
from typing import Any

from confluent_kafka import Consumer, KafkaException, Producer, TopicPartition

consumer_configs = {
    "bootstrap.servers": os.getenv("KAFKA_BOOTSTRAP_SERVERS"),
    "isolation.level": "read_committed",
    "auto.offset.reset": "latest",
    "group.id": os.getenv("KAFKA_CONSUMER_GROUP"),
    "enable.auto.commit": False,
    "enable.auto.offset.store": False,
}
producer_configs = {
    "bootstrap.servers": os.getenv("KAFKA_BOOTSTRAP_SERVERS"),
    "enable.idempotence": True,
    "transactional.id": os.getenv("KAFKA_CONSUMER_GROUP"),
}

consumer = Consumer(consumer_configs)
producer = Producer(producer_configs)


class KWPostAbortTransaction(Exception):
    ...


@dataclass
class KWMessage:
    topic: str
    value: str
    key: str


def send_eos_messages(
    messages: list[KWMessage],
    offsets: list[TopicPartition],
    group_metadata: Any,
) -> None:
    """Send Exactly-Once-Semantic messages with the transactional API."""

    producer.begin_transaction()
    try:
        for message in messages:
            producer.produce(message.topic, message.value, message.key)
    except (BufferError, KafkaException) as err:
        raise KWPostAbortTransaction() from err
    producer.send_offsets_to_transaction(offsets, group_metadata)
    while True:
        try:
            producer.commit_transaction(10.0)
            break
        except KafkaException as err:  # noqa
            if err.args[0].retriable():
                # retriable error, try again
                continue
            elif err.args[0].txn_requires_abort():
                producer.abort_transaction()
                raise KWPostAbortTransaction()
            else:
                # treat all other errors as fatal
                raise


def _decay_send_error():
    ...


def _increment_send_error():
    ...


# Send messages resulting from the processed message.
# This code is at the end of consumer's callback for its while-true-poll loop
...
next_offset = consumer.position(consumer.assignment())
try:
    send_eos_messages(out_messages, next_offset, consumer.consumer_group_metadata())
    _decay_send_error()
except KWPostAbortTransaction:
    # When a transaction has been aborted, rewind the consumer's offset. By
    # decrementing the offset, the current message will be processed again.
    next_offset[0].offset -= 1
    consumer.commit(offsets=next_offset)
    _increment_send_error()