FIX add error_cb to confluent.Consumer config in ConsumerFromTopic

[SuccessMoses]

closes: #43569

[rawwar]

should we try to use logger instead of a print here?

[SuccessMoses]

i think in other operators they use logger instead

[rawwar]

I think, the approach looks good and similar to what I have tested. Please update tests.

[mrk-andreev]

Looks promising!

error_cb(kafka.KafkaError): Callback for generic/global error events

We have to filter only related one ( https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#pythonclient-kafkaerror ).

[SuccessMoses]

Is this what you meant?

def error_callback(err):
    """Handle kafka errors."""
    if err.code() == KafkaError.NO_ERROR:
        return
    else:
        print("Exception received: ", err)
        raise KafkaAuthenticationError(f"Authentication failed: {err}")

[mrk-andreev]

I think that we would like to catch only Authentication issue while error_callback consume all errors (https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#pythonclient-kafkaerror).

[mrk-andreev]

It might not be the best practice to update user-provided data. As an alternative, we could create a shallow copy instead.

Additionally, we can assume that the user has already provided an error_cb. In this case, we need to decide how to handle this behavior, which is an open question. One possible solution is to call the user-defined error_cb first and then invoke our custom implementation if the user does not raise an exception. This approach makes sense because authentication errors are not recoverable, so they should interrupt execution.

[SuccessMoses]

Additionally, we can assume that the user has already provided an error_cb.

Is there a parameter for user defined error_cb?

[mrk-andreev]

Additionally, we can assume that the user has already provided an error_cb.

config may already contain handler. In this case you will override user defined handler.