Producer not publishing any message with topic using aioProducer in latest 2.12.0b1 version.

[jay-detroja3]

I have tried as per given example of asyncio. But it not working to create any topic and publish message in that.

Here is my code:

import asyncio
import logging

from confluent_kafka.aio import AIOProducer


logger = logging.getLogger(__name__)

async def error_cb(err):
    logger.error(f'Kafka error: {err}')

async def throttle_cb(event):
    logger.warning(f'Kafka throttle event: {event}')

async def stats_cb(stats_json_str):
    logger.info(f'Kafka stats: {stats_json_str}')


def configure_common(conf):
    bootstrap_servers = "localhost:9092"
    conf.update({
        'bootstrap.servers': bootstrap_servers,
        'logger': logger,
        'debug': 'conf',
        'error_cb': error_cb,
        'throttle_cb': throttle_cb,
        'stats_cb': stats_cb,
        'statistics.interval.ms': 5000,
    })

    return conf


async def start_producer():
    topic = "testtopic"
    producer = AIOProducer(configure_common(
        {
            'transactional.id': 'producer1'
        }), max_workers=5)

    await producer.init_transactions()
    transaction_active = False

    try:
        while True:
            await producer.begin_transaction()
            transaction_active = True

            produce_futures = [asyncio.create_task(producer.produce(topic=topic,
                                 key=f'testkey{i}',
                                 value=f'testvalue{i}'))for i in range(10)]

            results = await asyncio.gather(*produce_futures)

            for msg in results:
                logger.info(
                    'Produced to: {} [{}] @ {}'.format(msg.topic(),
                                                       msg.partition(),
                                                       msg.offset()))

            await producer.commit_transaction()
            transaction_active = False
            await asyncio.sleep(1)
    except Exception as e:
        logger.error(e)
    finally:
        if transaction_active:
            await producer.abort_transaction()
        await producer.stop()
        logger.info('Closed producer')

async def main():
    producer_task = asyncio.create_task(start_producer())
    await asyncio.gather(producer_task)

try:
    asyncio.run(main())
except asyncio.exceptions.CancelledError as e:
    logger.warning(f'Asyncio task was cancelled: {e}')

[MSeal]

Thanks for the report -- I think the asyncio example in examples that you based this on wasn't updated to the latest changes for the beta release.

So a couple issues in the example code setup here.

First is that the produce call needs to be awaited AND it returns a future after that point. Basically you are first awaiting the message is stored in the local queue, then you get a future for when it's delivered to the broker. This mirrors the user_callback pattern in synchronous producers.

The second is that awaiting those messages before flushing is causing deadlocks through making the async process waiting indefinitely. This I think is actually a bug, in that it should eventually flush without user prompt as we have a background task running to handle this. However it's easy enough to circumvent by adding an await flush before reading futures or reading those after transaction has been committed rather than before.

I cleaned up the example and posted it back with a working version. I'll also update the documented example in the repo to match as well as look into why automatic flush isn't firing in this situation.

import asyncio
import logging

from confluent_kafka.aio import AIOProducer


logger = logging.getLogger(__name__)

async def error_cb(err):
    logger.error(f'Kafka error: {err}')

async def throttle_cb(event):
    logger.warning(f'Kafka throttle event: {event}')

async def stats_cb(stats_json_str):
    logger.info(f'Kafka stats: {stats_json_str}')


def configure_common(conf):
    bootstrap_servers = "localhost:9092"
    conf.update({
        'bootstrap.servers': bootstrap_servers,
        'logger': logger,
        'debug': 'conf',
        'error_cb': error_cb,
        'throttle_cb': throttle_cb,
        'stats_cb': stats_cb,
        'statistics.interval.ms': 5000,
    })
    return conf


async def start_producer():
    topic = "testtopic"
    producer = AIOProducer(configure_common(
        {
            'transactional.id': 'producer1'
        }), max_workers=5)

    await producer.init_transactions()
    transaction_active = False

    try:
        while True:
            await producer.begin_transaction()
            transaction_active = True

            produce_futures = [await producer.produce(
                                 topic=topic,
                                 key=f'testkey{i}',
                                 value=f'testvalue{i}')
                               for i in range(10)]
            print(f"Produced {len(produce_futures)} messages")

            # If you put the message reading after the transaction commit you don't need this flush call
            await producer.flush()
            print("Waiting for messages to be produced")
            for msg in await asyncio.gather(*produce_futures):
                print(
                    'Produced to: {} [{}] @ {}'.format(msg.topic(),
                                                       msg.partition(),
                                                       msg.offset()))

            await producer.commit_transaction()

            transaction_active = False
            await asyncio.sleep(1)

    except Exception as e:
        logger.error(e)
    finally:
        if transaction_active:
            await producer.abort_transaction()
        await producer.close()
        logger.info('Closed producer')

async def main():
    await start_producer()

try:
    asyncio.run(main())
except asyncio.exceptions.CancelledError as e:
    logger.warning(f'Asyncio task was cancelled: {e}')

[jay-detroja3]

Okay Great It working fine. Thank you for support.