ISV-6424: Implement base producer (#4)

* add integration tests
* fix out-of-order offset commits

Signed-off-by: Marek Szymutko <mszymutk@redhat.com>
Assisted-by: Claude-4.5-Sonnet (Cursor)

Author: BorekZnovustvoritel

.github/workflows/integration.yml

@@ -0,0 +1,53 @@
+name: Integration Tests
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+jobs:
+  integration:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.13"]
+    
+    steps:
+    - uses: actions/checkout@v4
+    
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+    
+    - name: Install uv
+      uses: astral-sh/setup-uv@v5
+      with:
+        version: "latest"
+    
+    - name: Install tox
+      run: |
+        uv venv
+        uv pip install tox
+    
+    - name: Start Kafka
+      run: |
+        docker compose up -d
+        echo "Waiting for Kafka to be healthy..."
+        timeout 120 bash -c 'until docker compose ps kafka | grep -q "healthy"; do sleep 2; done'
+        echo "Kafka is healthy and ready for tests"
+    
+    - name: Run integration tests
+      run: |
+        uv run tox -e integration
+    
+    - name: Show Kafka logs on failure
+      if: failure()
+      run: |
+        docker compose logs kafka
+    
+    - name: Cleanup
+      if: always()
+      run: |
+        docker compose down -v

README.md

@@ -11,3 +11,46 @@ used.
 
 The aim is to provide a fault-tolerant platform for parallel message
 processing.
+
+## Local testing
+
+This project uses [`uv`][1]. To set up the project locally, use
+
+```bash
+uv pip install .
+```
+
+To test, you also need development tools, add them to your local environment
+using this command:
+
+```bash
+uv pip install --group dev
+```
+
+Then you can use [`tox`][2] to run linting and unit tests.
+
+For integration tests you also need [`podman`][3] or [`docker`][4] with
+`compose`. Run:
+
+```bash
+docker compose up -d 
+```
+
+Wait a while and then run:
+
+```bash
+tox -e integration
+```
+
+Don't forget to clean up the test environment afterward. Use
+
+```bash
+podman compose down
+```
+
+to do that (or switch `podman` with `docker` depending on your tool of choice).
+
+[1]: https://docs.astral.sh/uv/
+[2]: https://tox.wiki/en/4.32.0/
+[3]: https://podman.io/
+[4]: https://www.docker.com/

compose.yml

@@ -0,0 +1,52 @@
+services:
+  kafka:
+    image: apache/kafka:latest
+    hostname: kafka
+    container_name: kafka
+    ports:
+      - "9092:9092"
+      - "9093:9093"
+    environment:
+      # KRaft settings
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9094
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      
+      # Listener configuration
+      KAFKA_LISTENERS: SASL_PLAINTEXT://0.0.0.0:9092,PLAINTEXT://0.0.0.0:9093,CONTROLLER://0.0.0.0:9094
+      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://localhost:9092,PLAINTEXT://localhost:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: SASL_PLAINTEXT:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      
+      # SASL/SCRAM configuration
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-512
+      KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-512
+      
+      # Dynamic topic creation
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+      
+      # Compression settings
+      KAFKA_COMPRESSION_TYPE: snappy
+      
+      # Log settings
+      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      
+      # JAAS configuration for SASL
+      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
+      
+    volumes:
+      - ./kafka_config/kafka-startup.sh:/tmp/kafka-startup.sh:ro,z
+      - ./kafka_config/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf:ro,z
+    tmpfs:
+      - /tmp/kraft-combined-logs:uid=1000,gid=1000
+    command: ["/bin/bash", "/tmp/kafka-startup.sh"]
+    healthcheck:
+      test: ["CMD", "/opt/kafka/bin/kafka-broker-api-versions.sh", "--bootstrap-server", "localhost:9093"]
+      interval: 10s
+      timeout: 10s
+      retries: 5
+      start_period: 30s

kafka_config/kafka-startup.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+set -e
+
+echo "Starting Kafka in background..."
+/etc/kafka/docker/run &
+KAFKA_PID=$!
+
+# Wait for Kafka to be ready
+echo "Waiting for Kafka to be ready..."
+for i in {1..60}; do
+  if /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9093 > /dev/null 2>&1; then
+    echo "Kafka is ready!"
+    break
+  fi
+  if [ $i -eq 60 ]; then
+    echo "Kafka failed to start in time"
+    exit 1
+  fi
+  sleep 2
+done
+
+echo "Creating SCRAM credentials..."
+
+# Create SCRAM credentials for test user
+/opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9093 \
+  --alter \
+  --add-config 'SCRAM-SHA-512=[password=testpassword]' \
+  --entity-type users \
+  --entity-name testuser
+
+echo "================================"
+echo "SCRAM credentials created successfully!"
+echo "Username: testuser"
+echo "Password: testpassword"
+echo "================================"
+
+# Keep container running
+wait $KAFKA_PID
+
+

kafka_config/kafka_server_jaas.conf

@@ -0,0 +1,4 @@
+KafkaServer {
+    org.apache.kafka.common.security.scram.ScramLoginModule required;
+};
+

pyproject.toml

@@ -23,11 +23,13 @@ exclude = ["tests*"]
 
 [tool.pytest.ini_options]
 pythonpath = ["src"]
+
 [dependency-groups]
 dev = [
     "coverage>=7.12.0",
     "mypy>=1.18.2",
     "pytest>=9.0.1",
+    "pytest-asyncio>=0.24.0",
     "pytest-cov>=7.0.0",
     "ruff>=0.14.5",
     "tox>=4.32.0",

src/retriable_kafka_client/__init__.py

@@ -1,7 +1,7 @@
 """Retriable Kafka client module"""
 
 from .consumer import BaseConsumer
-from .types import TopicConfig
+from .types import ConsumerConfig
 from .orchestrate import consume_topics
 
-__all__ = ("BaseConsumer", "TopicConfig", "consume_topics")
+__all__ = ("BaseConsumer", "ConsumerConfig", "consume_topics")