Fix no-payload commit in consumer

The confluent-kafka `Consumer.commit` method can be called with no
parameters, then the current partition assignment's offsets are used
(see the documentation at
https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.Consumer.commit).

The previous behaviour was erroneous, raising a `TypeError`, thus an
HTTP 500 when in the REST proxy's consumer manager a commit is performed
with an empty payload.

karapace/kafka/consumer.py

@@ -86,7 +86,10 @@ def commit(  # type: ignore[override]
             if message is not None:
                 return super().commit(message=message, asynchronous=False)
 
-            return super().commit(offsets=offsets, asynchronous=False)
+            if offsets is not None:
+                return super().commit(offsets=offsets, asynchronous=False)
+
+            return super().commit(asynchronous=False)
         except KafkaException as exc:
             raise_from_kafkaexception(exc)
 

tests/integration/kafka/test_consumer.py

@@ -137,6 +137,31 @@ def test_commit_offsets(
         assert committed_partition.partition == 0
         assert committed_partition.offset == 2
 
+    def test_commit_offsets_empty(
+        self,
+        producer: KafkaProducer,
+        consumer: KafkaConsumer,
+        new_topic: NewTopic,
+    ) -> None:
+        consumer.subscribe([new_topic.topic])
+        first_fut = producer.send(new_topic.topic)
+        second_fut = producer.send(new_topic.topic)
+        producer.flush()
+        first_fut.result()
+        second_fut.result()
+        consumer.poll(timeout=POLL_TIMEOUT_S)
+        consumer.poll(timeout=POLL_TIMEOUT_S)
+
+        [topic_partition] = consumer.commit(offsets=None, message=None)  # default parameters
+        [committed_partition] = consumer.committed([TopicPartition(new_topic.topic, partition=0)])
+
+        assert topic_partition.topic == new_topic.topic
+        assert topic_partition.partition == 0
+        assert topic_partition.offset == 2
+        assert committed_partition.topic == new_topic.topic
+        assert committed_partition.partition == 0
+        assert committed_partition.offset == 2
+
     def test_commit_raises_for_unknown_partition(
         self,
         consumer: KafkaConsumer,