Unable to commit offsets explicitly when they contain duplicates.

[Iamcerba]

Description

Whenever you try to commit offset explicitly, use synchronous request, and the "offsets" list contains duplicates for the partition in the topic then the following exception is being thrown:

cimpl.KafkaException: KafkaError{code=NETWORK_EXCEPTION,val=13,str="Commit failed: Broker: Broker disconnected before response received"}

With async everything works fine.

How to reproduce

offsets = list()

for i in range(1, 10):
    offsets.append(TopicPartition(topic='topic', partition=0, offset=random.randint(1, 100))

consumer.commit(offsets=offsets, asynchronous=False)

confluent-kafka-python: ('1.5.0', 17104896)
librdkafka: ('1.5.0', 17105151)
Apache Kafka broker version: Azure EventHubs
Consumer conf:

conf={
            'bootstrap.servers': ...,
            'client.id': ...,
            'group.id': ...,
            'default.topic.config': {
                'auto.offset.reset': 'smallest'
            },
            'enable.auto.commit': 'False',
            'security.protocol': 'SASL_SSL',
            'sasl.mechanism': 'PLAIN',
            'sasl.username': ...,
            'sasl.password': ...
        }

Operating system: ubuntu

Expected behavior

I know that the problem initially on the client-side but it will be useful to be more explicit in the error response.

[mhowlett]

yes, that is a surprising error message - you shouldn't be getting a disconnect error due to this.