# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()


from enum import Enum
from collections import namedtuple, defaultdict, deque
from aiokafka.structs import TopicPartition
from aiokafka.util import create_future
        return self._pid_and_epoch.pid

    @property
    def producer_epoch(self):
        return self._pid_and_epoch.epoch

    # TRANSACTION PART

    def _transition_to(self, target):
        assert TransactionState.is_transition_valid(self.state, target), \
            f"Invalid state transition {self.state} -> {target}"
        self.state = target

    def begin_transaction(self):
        self._transition_to(TransactionState.IN_TRANSACTION)
        self._transaction_waiter = create_future()

    def committing_transaction(self):
        if self.state == TransactionState.ABORTABLE_ERROR:
            # Raise error to user, we can only abort at this point
            self._transaction_waiter.result()

        self._transition_to(TransactionState.COMMITTING_TRANSACTION)
        self.notify_task_waiter()

    def aborting_transaction(self):
        self._transition_to(TransactionState.ABORTING_TRANSACTION)

        # If we had an abortable error we need to create a new waiter
        if self._transaction_waiter.done():

# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()


# Path: aiokafka/producer/transaction_manager.py
from enum import Enum
from collections import namedtuple, defaultdict, deque
from aiokafka.structs import TopicPartition
from aiokafka.util import create_future

        return self._pid_and_epoch.pid

    @property
    def producer_epoch(self):
        return self._pid_and_epoch.epoch

    # TRANSACTION PART

    def _transition_to(self, target):
        assert TransactionState.is_transition_valid(self.state, target), \
            f"Invalid state transition {self.state} -> {target}"
        self.state = target

    def begin_transaction(self):
        self._transition_to(TransactionState.IN_TRANSACTION)
        self._transaction_waiter = create_future()

    def committing_transaction(self):
        if self.state == TransactionState.ABORTABLE_ERROR:
            # Raise error to user, we can only abort at this point
            self._transaction_waiter.result()

        self._transition_to(TransactionState.COMMITTING_TRANSACTION)
        self.notify_task_waiter()

    def aborting_transaction(self):
        self._transition_to(TransactionState.ABORTING_TRANSACTION)

        # If we had an abortable error we need to create a new waiter
        if self._transaction_waiter.done():

            self._transaction_waiter = create_future()