Merge pull request #2 from shafa-dev/kafka-rest-proxy

add Confluent REST Proxy api

Author: bogdan-zs

README.md

@@ -1,5 +1,5 @@
 # kafka-bridge-client
-Python async client for [Strimzi Kafka Bridge](https://github.com/strimzi/strimzi-kafka-bridge). Package include consumer only.
+Python async client for [Strimzi Kafka Bridge](https://github.com/strimzi/strimzi-kafka-bridge) and [Confluent REST Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html) Package include consumer only.
 
 [![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT)
 [![PRs Welcome](https://img.shields.io/badge/PRs-welcome-green.svg)](https://github.com/shafa-dev/kafka-bridge-client/issues)
@@ -11,11 +11,13 @@ pip install kafka-bridge-client
 ```
 
 ## Usage
+By default client use [Strimzi Kafka Bridge](https://github.com/strimzi/strimzi-kafka-bridge) API
 ```python
 from kafka_bridge_client import KafkaBridgeConsumer
 
+# Strimzi Kafka Bridge
 
-consumer = KafkaBridgeConsumer(
+consumer1 = KafkaBridgeConsumer(
     'topic1',
     'topic2',
     group_id='my-group,
@@ -25,7 +27,19 @@ consumer = KafkaBridgeConsumer(
     consumer_name='consumer-name',
 )
 
-async for rec in consumer.get_records():
+# Confluent REST Proxy
+consumer2 = KafkaBridgeConsumer(
+    'topic1',
+    'topic2',
+    group_id='my-group,
+    auto_offset_reset='earliest',
+    enable_auto_commit=False,
+    bootstrap_server='your-kafka-bridge-url',
+    consumer_name='consumer-name',
+    proxy='confluent'
+)
+
+async for rec in consumer1.get_records():
     print(rec['value'])
     await consumer.commit()
 ```

kafka_bridge_client/consumer.py

@@ -57,17 +57,27 @@ def __init__(
         sleep_interval_seconds: int = 2,
         client_timeout_seconds: int = 15,
         headers: t.Dict[str, t.Any] = None,
+        proxy: t.Literal['strimzi', 'confluent'] = 'strimzi'
     ) -> None:
         self._group_id = group_id
         self._consumer_name = consumer_name
         self._topics = topics
         self._offsets: t.Dict[str, t.Dict[str, t.Any]] = {}
-        self._config = {
-            'auto.offset.reset': auto_offset_reset,
-            'enable.auto.commit': enable_auto_commit,
-            'format': 'binary',
-            'name': consumer_name,
-        }
+        if proxy == 'strimzi':
+            self._config = {
+                'auto.offset.reset': auto_offset_reset,
+                'enable.auto.commit': enable_auto_commit,
+                'format': 'binary',
+                'name': consumer_name,
+            }
+        elif proxy == 'confluent':
+            self._config = {
+                'auto.offset.reset': auto_offset_reset,
+                'auto.commit.enable': 'true' if enable_auto_commit else 'false',
+                'format': 'binary',
+                'name': consumer_name,
+            }
+
         # Delay between fetching of records if
         # the previous fetch return zero records
         self._sleep_interval_seconds = sleep_interval_seconds

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "kafka-bridge-client"
-version = "0.1.3"
+version = "0.1.4"
 description = "Python client for Strimzi Kafka Bridge"
 authors = ["Bogdan Zaseka <zaseka.bogdan@gmail.com>"]
 license = "MIT"