SystemError if assign_cb raises Exception

[coldeasy]

Description

A SystemError exception is raised when calling poll or close on a consumer if the on_assign callback raises an Exception.

    def log(self, level, msg, *args, **kwargs):
        """
        Log 'msg % args' with the integer severity 'level'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.log(level, "We have a %s", "mysterious problem", exc_info=1)
        """
>       if not isinstance(level, int):
E       SystemError: <built-in function isinstance> returned a result with an error set

I think this is due to the cpython code not clearing the python exception during callbacks and subsequent callbacks executing python code.
Reading the docs, I get the impression that the librdkafka callbacks which execute python code should clear exceptions PyErr_Clear and restore them when python code is calling cpython functions (e.g. check and set the error when consumer.poll is called). If this is right, we could probably replace the crashed counter with the exception info.

How to reproduce

    import logging
    logging.basicConfig(level=logging.DEBUG)
    from confluent_kafka import Producer
    from confluent_kafka import Consumer
    prod = Producer(
        {
            'bootstrap.servers': servers,
            'debug': 'all'
        },
        logger=logging.getLogger('')
    )
    prod.produce('test', b'data')
    assert prod.flush(10) == 0
    cons = Consumer(
        {
            'bootstrap.servers': servers,
            'group.id': 'test',
            'auto.offset.reset': 'smallest',
            'debug': 'all'
        },
        logger=logging.getLogger('')
    )

    def _on_assign(self, *args, **kwargs):
        raise Exception("Oops")

    cons.subscribe(['test'], on_assign=_on_assign, on_revoke=_on_assign)
    msg = cons.poll(10)
    assert msg is not None
    cons.close()

I can provide the debug logs if necessary.

Checklist

Please provide the following information:

• confluent_kafka.version(), confluent_kafka.libversion() == (('1.2.2', 16908800), ('1.2.2', 16909055))
• 0.10.2.1
• Client configuration: {...}
• Operating system:
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[coldeasy]

This is the stack trace that's printed to stderr

self = <cimpl.Consumer object at 0x7efe1e712598>, args = ([TopicPartition{topic=test,partition=0,offset=-1001,error=None}],), kwargs = {}

    def _on_assign(self, *args, **kwargs):
>       raise Exception("Oops")
E       Exception: Oops

test_kafka_subscriber.py:304: Exception

The above exception was the direct cause of the following exception:

self = <RootLogger root (WARNING)>, level = 10, msg = '%s [%s] %s'
args = ('HEARTBEAT', 'rdkafka#consumer-2', '[thrd:main]: GroupCoordinator/0: Heartbeat for group "test" generation id 1'), kwargs = {}

    def log(self, level, msg, *args, **kwargs):
        """
        Log 'msg % args' with the integer severity 'level'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.log(level, "We have a %s", "mysterious problem", exc_info=1)
        """
>       if not isinstance(level, int):
E       SystemError: <built-in function isinstance> returned a result with an error set

/usr/local/lib/python3.6/logging/__init__.py:1368: SystemError

The part below The above exception was the direct cause of the following exception: is repeated several times

[olejorgenb]

Just mentioning a scenario where this is annoying: (copied from #150 (comment))

This seems to happen when the process is terminated using SIGTERM and a signal handler is installed to raise sys.exit(0) (raises SystemExit). Probably something similar happen for SIGINT.

We frequently observer: SystemError: <built-in method consume of cimpl.Consumer object at 0x7fe3858d8d60> returned NULL without setting an exception raised from the consume method when this happen. The active frame at signal time is our on_revoke callback, so I assume this issue is the root case.

In our case there's no real harm - other than cluttering our error reports.