# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")


import asyncio
import os
import async_timeout
from asyncio import AbstractEventLoop
from typing import Awaitable, Dict, Tuple, TypeVar, Union, cast
from packaging.version import Version
from .structs import OffsetAndMetadata, TopicPartition
    offsets: Dict[TopicPartition, Union[int, Tuple[int, str], OffsetAndMetadata]]
) -> Dict[TopicPartition, OffsetAndMetadata]:
    # validate `offsets` structure
    if not offsets or not isinstance(offsets, dict):
        raise ValueError(offsets)

    formatted_offsets = {}
    for tp, offset_and_metadata in offsets.items():
        if not isinstance(tp, TopicPartition):
            raise ValueError("Key should be TopicPartition instance")

        if isinstance(offset_and_metadata, int):
            offset, metadata = offset_and_metadata, ""
        else:
            try:
                offset, metadata = offset_and_metadata
            except Exception:
                raise ValueError(offsets)

            if not isinstance(metadata, str):
                raise ValueError("Metadata should be a string")

        formatted_offsets[tp] = OffsetAndMetadata(offset, metadata)
    return formatted_offsets


def get_running_loop() -> asyncio.AbstractEventLoop:
    loop = asyncio.get_event_loop()
    if not loop.is_running():
        raise RuntimeError(

# 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
import asyncio
import os
import async_timeout
from asyncio import AbstractEventLoop
from typing import Awaitable, Dict, Tuple, TypeVar, Union, cast
from packaging.version import Version
from .structs import OffsetAndMetadata, TopicPartition

    offsets: Dict[TopicPartition, Union[int, Tuple[int, str], OffsetAndMetadata]]
) -> Dict[TopicPartition, OffsetAndMetadata]:
    # validate `offsets` structure
    if not offsets or not isinstance(offsets, dict):
        raise ValueError(offsets)

    formatted_offsets = {}
    for tp, offset_and_metadata in offsets.items():
        if not isinstance(tp, TopicPartition):
            raise ValueError("Key should be TopicPartition instance")

        if isinstance(offset_and_metadata, int):
            offset, metadata = offset_and_metadata, ""
        else:
            try:
                offset, metadata = offset_and_metadata
            except Exception:
                raise ValueError(offsets)

            if not isinstance(metadata, str):
                raise ValueError("Metadata should be a string")

        formatted_offsets[tp] = OffsetAndMetadata(offset, metadata)
    return formatted_offsets


def get_running_loop() -> asyncio.AbstractEventLoop:
    loop = asyncio.get_event_loop()
    if not loop.is_running():
        raise RuntimeError(

            "The object should be created within an async function or "