# 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
    loop = get_running_loop()
    return loop.create_task(coro)


def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
    if loop is None:
        loop = get_running_loop()
    return loop.create_future()


async def wait_for(fut: Awaitable[T], timeout: Union[None, int, float] = None) -> T:
    # A replacement for buggy (since 3.8.6) `asyncio.wait_for()`
    # https://bugs.python.org/issue42130
    async with async_timeout.timeout(timeout):
        return await fut


def parse_kafka_version(api_version: str) -> Tuple[int, int, int]:
    parsed = Version(api_version).release
    if not 2 <= len(parsed) <= 3:
        raise ValueError(api_version)
    version = cast(Tuple[int, int, int], (parsed + (0,))[:3])

    if not (0, 9) <= version < (3, 0):
        raise ValueError(api_version)
    return version


def commit_structure_validate(
    offsets: Dict[TopicPartition, Union[int, Tuple[int, str], OffsetAndMetadata]]

# 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

    loop = get_running_loop()
    return loop.create_task(coro)


def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
    if loop is None:
        loop = get_running_loop()
    return loop.create_future()


async def wait_for(fut: Awaitable[T], timeout: Union[None, int, float] = None) -> T:
    # A replacement for buggy (since 3.8.6) `asyncio.wait_for()`
    # https://bugs.python.org/issue42130
    async with async_timeout.timeout(timeout):
        return await fut


def parse_kafka_version(api_version: str) -> Tuple[int, int, int]:
    parsed = Version(api_version).release
    if not 2 <= len(parsed) <= 3:
        raise ValueError(api_version)
    version = cast(Tuple[int, int, int], (parsed + (0,))[:3])

    if not (0, 9) <= version < (3, 0):
        raise ValueError(api_version)
    return version


def commit_structure_validate(
    offsets: Dict[TopicPartition, Union[int, Tuple[int, str], OffsetAndMetadata]]

) -> Dict[TopicPartition, OffsetAndMetadata]: