Not able to read Kafka Messages with Glassflow

[Dhairya1007]

I had a problem while using Glassflow to connect to a Kafka Stream from Confluent Cloud. After all the configurations and the pipeline deployment, the pipeline was not able to ingest any incoming messages into ClickHouse.

More specifically, I can see an error message in establishing the connectivity with Kafka and then a message saying "No messages to send".

Any help with this would be greatly appreciated.

[PabloPardoGarcia]

Hi @Dhairya1007 thanks for reaching out. Could you provide the following information so we can debug the issue?

• Clickhouse ETL version
• Application logs
• Kafka Connection parameters (without sensitive information)

Thanks

[Dhairya1007]

Sure @PabloPardoGarcia. Here is the requested information:

Clickhouse ETL version: 0.1.2
Application logs:

{"time":"2025-05-21T14:04:49.281578013Z","level":"INFO","msg":"HTTP server listening","Addr":":8080"}
{"time":"2025-05-21T14:04:53.751512126Z","level":"INFO","msg":"request","component":"glassflow_api","latency":520583,"remote_ip":"172.20.0.7:59148","host":"app:8080","method":"GET","path":"/api/v1/pipeline","referer":"","user_agent":"python-httpx/0.28.1","status":404}
{"time":"2025-05-21T14:04:54.881544668Z","level":"INFO","msg":"[Kafka:user_events_with_duplicates to JS] [INF] starting NATS-Kafka Bridge","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:04:54.94057021Z","level":"DEBUG","msg":"Insert query","pipeline_id":"deduplication-demo-pipeline","component":"clickhouse_sink","query":"INSERT INTO default.user_events_deduplicated (event_id, user_id, created_at, user_name, user_email)"}
{"time":"2025-05-21T14:04:54.97434846Z","level":"INFO","msg":"request","component":"glassflow_api","latency":182442583,"remote_ip":"172.20.0.7:59160","host":"app:8080","method":"POST","path":"/api/v1/pipeline","referer":"","user_agent":"python-httpx/0.28.1","status":200}
{"time":"2025-05-21T14:04:54.975751876Z","level":"INFO","msg":"ClickHouse sink started","pipeline_id":"deduplication-demo-pipeline","component":"clickhouse_sink"}
{"time":"2025-05-21T14:05:16.105160845Z","level":"INFO","msg":"[Kafka:user_events_with_duplicates to JS] [INF] error starting Kafka:user_events_with_duplicates to JetStream:gf-stream-user_events_with_duplicates-ba27ed27-546d-4543-935a-4f269b3d4805.input, failed to create consumer: new kafka consumer group: kafka: client has run out of available brokers to talk to: unexpected EOF","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:05:16.105663553Z","level":"ERROR","msg":"[Kafka:user_events_with_duplicates to JS] [ERR] error starting bridge, start connectors: failed to create consumer: new kafka consumer group: kafka: client has run out of available brokers to talk to: unexpected EOF","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:05:16.105696178Z","level":"INFO","msg":"[Kafka:user_events_with_duplicates to JS] [INF] stopping bridge","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:05:16.105731595Z","level":"INFO","msg":"[Kafka:user_events_with_duplicates to JS] [INF] shutting down connection Kafka:user_events_with_duplicates to JetStream:gf-stream-user_events_with_duplicates-ba27ed27-546d-4543-935a-4f269b3d4805.input","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:05:16.105750261Z","level":"INFO","msg":"[Kafka:user_events_with_duplicates to JS] [INF] disconnected from NATS","pipeline_id":"deduplication-demo-pipeline","component":"kafka_bridge"}
{"time":"2025-05-21T14:05:25.06248646Z","level":"DEBUG","msg":"No messages to send","pipeline_id":"deduplication-demo-pipeline","component":"clickhouse_sink"}
{"time":"2025-05-21T14:05:55.176295168Z","level":"DEBUG","msg":"No messages to send","pipeline_id":"deduplication-demo-pipeline","component":"clickhouse_sink"}

Connection parameters:

"connection_params": {
        "brokers": [
          "<broker-address>:9092"
        ],
        "protocol": "SASL_SSL",
        "mechanism": "PLAIN",
        "username": "<user>",
        "password": "<password>",
        "skip_auth": false
      },

[PabloPardoGarcia]

Looks like this bug was patched in version 0.1.3.
Could you try to run the same pipeline with the latest version?

[Dhairya1007]

Thank you for your help @PabloPardoGarcia . I was able to successfully run my pipeline after updating to the latest version.