How to design Kafka topics and kafkareceiver config for OTel → Kafka → Flink → Kafka → ClickHouse pipeline?

[mitht]

I have the following pipeline:

OTel Collector → Kafka (otlp_spans) → Flink → Kafka (processed_otlp_spans) → OTel Collector → ClickHouse

I'm confused about:

1. Which topic Flink should read/write
2. Whether I need multiple kafka receivers
3. How to configure kafkareceiver correctly

Do I need two OpenTelemetry Collectors in this pipeline?
collector-a

receivers:
  otlp:
    protocols:
      grpc:
      http:

processors:
  batch:
    timeout: 5s
    send_batch_size: 1024

exporters:
  kafka:
    brokers:
      - localhost:9092
    topic: otlp_spans   
    encoding: otlp_proto

    auth:
      sasl:
        username: "user"
        password: "secret"
        mechanism: "SCRAM-SHA-512"

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [kafka]

collector-b

receivers:
  kafka:
    brokers:
      - localhost:9092
    topics:
      - processed_otlp_spans   # ✅ Flink output
    encoding: otlp_proto
    group_id: otel-clickhouse-consumer

    auth:
      sasl:
        username: "user"
        password: "secret"
        mechanism: "SCRAM-SHA-512"

processors:
  batch:
    timeout: 5s
    send_batch_size: 2048

exporters:
  clickhouse:
    endpoint: tcp://localhost:9000?compress=lz4
    database: otel

    traces_table_name: otel_traces
    logs_table_name: otel_logs
    metrics_table_name: otel_metrics

    timeout: 10s

    create_schema: true

    sending_queue:
      enabled: true
      num_consumers: 4
      queue_size: 10000

service:
  pipelines:
    traces:
      receivers: [kafka]
      processors: [batch]
      exporters: [clickhouse]

    metrics:
      receivers: [kafka]
      processors: [batch]
      exporters: [clickhouse]

    logs:
      receivers: [kafka]
      processors: [batch]
      exporters: [clickhouse]

java-flink code

byte[] data = record.value();
ExportTraceServiceRequest request =
    ExportTraceServiceRequest.parseFrom(data);
request.getResourceSpansList().forEach(resourceSpans -> {
    resourceSpans.getScopeSpansList().forEach(scopeSpans -> {
        scopeSpans.getSpansList().forEach(span -> {

            String traceId = bytesToHex(span.getTraceId().toByteArray());
            String spanId = bytesToHex(span.getSpanId().toByteArray());
            String name = span.getName();

            System.out.println("traceId=" + traceId);
            System.out.println("span=" + name);
        });
    });
});

byte[] output = request.toByteArray();

producer.send(new ProducerRecord<>(
    "processed_otlp_spans",
    output
));

[s3onghyun]

Two Collectors is the right shape here, and your configs are basically correct — let me map them to your pipeline and flag the one thing that usually breaks it.

Topic flow

• collector-a: otlp receiver → kafka exporter topic: otlp_spans
• Flink: consumes otlp_spans, produces processed_otlp_spans
• collector-b: kafka receiver topics: [processed_otlp_spans] → clickhouse exporter

So Flink reads otlp_spans and writes processed_otlp_spans; collector-a and collector-b never share a topic. That's exactly what you have.

1. Which topic does Flink read/write — reads otlp_spans, writes processed_otlp_spans. Correct as drawn.

2. Multiple kafka receivers? — No. One receiver consuming one topic (or a list) is enough. You'd only add a second receiver if you consumed another, differently-encoded topic.

3. Two Collectors? — Yes. The data leaves to Kafka, gets transformed by Flink, and comes back on a different topic; a single Collector can't pause mid-pipeline for Flink, so a-then-b is correct.

The thing that actually bites: encoding must match what Flink writes.
collector-b's encoding (default otlp_proto) has to match the serialization Flink produces on processed_otlp_spans. If Flink re-emits OTLP protobuf, leave it as otlp_proto. If Flink writes JSON or a custom schema, the receiver fails to unmarshal and nothing reaches ClickHouse — set encoding to match (e.g. otlp_json) or have Flink emit OTLP proto. This is the most common cause of "collector-b consumes but ClickHouse stays empty."

Config-key version note (worth pinning):

• On the receiver, topic (singular) was deprecated in v0.142.0 in favor of topics (a list) — you're already on topics, good.
• Recent contrib also added signal-specific keys (traces::topic, traces::encoding, …) that deprecate the generic topic/encoding. On a current build prefer those; otherwise the generic keys still work.

If span ordering matters for your Flink logic: spans of one trace can land on different partitions of otlp_spans. If Flink does per-trace aggregation, key by trace ID on the producer side (the kafka exporter can partition by trace ID) so a whole trace stays on one partition.

Net: the two-Collector design is right — the field to double-check is collector-b's encoding vs Flink's actual output format.