Add High-Throughput Kafka Pipeline

Author: ilyario

docs/en/architecture.md

@@ -55,6 +55,7 @@ Transformations are applied in order: `timestamp`, `flatten`, `filter`, `mask`,
   - **Resources**: optional CPU/memory for the processor pod.
   - **Scheduling**: optional `nodeSelector`, `affinity`, `tolerations`.
   - **CheckpointPersistence**: optional; defaults to `true`. When enabled, polling sources (PostgreSQL, ClickHouse, Trino) persist read position to a ConfigMap, reducing duplicates on restart. Set to `false` to disable.
+  - **ChannelBufferSize**: optional; defaults to `100`. Buffer size for message channels between source, processor, and sink. Use 500–1000 for high Kafka throughput to reduce blocking when the sink is slower than the source.
 
 Secrets can be referenced via `SecretRef` in the spec; the operator resolves them before writing the spec into the ConfigMap.
 
@@ -77,6 +78,7 @@ flowchart TB
     Resources["resources (optional)"]
     Scheduling["scheduling (optional)"]
     Checkpoint["checkpointPersistence (optional)"]
+    ChannelBuffer["channelBufferSize (optional)"]
     Image["processorImage / processorVersion (optional)"]
   end
 
@@ -91,6 +93,7 @@ flowchart TB
   Spec --> Resources
   Spec --> Scheduling
   Spec --> Checkpoint
+  Spec --> ChannelBuffer
   Spec --> Image
 ```
 

docs/en/connectors.md

@@ -905,3 +905,38 @@ DataFlow Operator supports configuring a separate sink for messages that failed
 
 For configuration, error message structure, and error types, see [Error Handling](errors.md).
 
+## Performance Recommendations
+
+### Spec-level settings
+
+- **channelBufferSize** (default 100): Buffer size for message channels between source, processor, and sink. For high Kafka throughput (tens of thousands msg/s), increase to 500–1000 to reduce blocking when the sink is slower than the source.
+
+### Kafka
+
+- Use multiple brokers for fault tolerance
+- Configure an appropriate consumer group size for parallel processing
+- Use batch writes for higher throughput
+
+### PostgreSQL
+
+- Increase `batchSize` for the sink (recommended 50–100)
+- Add indexes on frequently queried columns
+- Tune `pollInterval` based on data update frequency
+
+## Troubleshooting
+
+### Connection issues
+
+1. Verify data source accessibility from the cluster
+2. Ensure credentials are correct
+3. Check Kubernetes network policies
+4. For TLS, verify certificates
+
+### Performance issues
+
+1. Increase `channelBufferSize` (500–1000) for high Kafka load
+2. Increase batch sizes for sinks
+3. Tune `pollInterval` for sources
+4. Scale operator instances if needed
+5. Monitor message processing metrics
+

docs/en/examples.md

@@ -240,6 +240,28 @@ kubectl apply -f dataflow/config/samples/kafka-to-postgres-secrets.yaml
 
 For supported fields, TLS certificates, and troubleshooting, see [Using Kubernetes Secrets](connectors.md#using-kubernetes-secrets).
 
+## High-Throughput Kafka Pipeline
+
+For high Kafka message rates (tens of thousands msg/s), increase `channelBufferSize` and sink `batchSize`:
+
+```yaml
+spec:
+  channelBufferSize: 500   # default 100; reduces blocking when sink is slower than source
+  source:
+    type: kafka
+    config:
+      brokers: [localhost:9092]
+      topic: high-volume-topic
+      consumerGroup: dataflow-group
+  sink:
+    type: postgresql
+    config:
+      connectionString: "..."
+      table: events
+      batchSize: 500
+      batchFlushIntervalSeconds: 2
+```
+
 ## Configuring Pod Resources and Placement
 
 Each DataFlow resource creates a separate pod (Deployment) for processing. You can configure resources, node selection, affinity, and tolerations for these pods.

docs/ru/architecture.md

@@ -55,6 +55,7 @@ flowchart LR
   - **Resources**: опциональные CPU/память для пода процессора.
   - **Scheduling**: опционально `nodeSelector`, `affinity`, `tolerations`.
   - **CheckpointPersistence**: опционально; по умолчанию `true`. При включении polling-источники (PostgreSQL, ClickHouse, Trino) сохраняют позицию чтения в ConfigMap, уменьшая дубликаты при перезапуске. Задайте `false` для отключения.
+  - **ChannelBufferSize**: опционально; по умолчанию `100`. Размер буфера каналов между source, processor и sink. Используйте 500–1000 при высокой нагрузке Kafka, чтобы снизить блокировки, когда sink медленнее source.
 
 Секреты задаются через `SecretRef` в spec; оператор подставляет их перед записью spec в ConfigMap.
 
@@ -77,6 +78,7 @@ flowchart TB
     Resources["resources (опционально)"]
     Scheduling["scheduling (опционально)"]
     Checkpoint["checkpointPersistence (опционально)"]
+    ChannelBuffer["channelBufferSize (опционально)"]
     Image["processorImage / processorVersion (опционально)"]
   end
 
@@ -91,6 +93,7 @@ flowchart TB
   Spec --> Resources
   Spec --> Scheduling
   Spec --> Checkpoint
+  Spec --> ChannelBuffer
   Spec --> Image
 ```
 

docs/ru/connectors.md

@@ -1290,6 +1290,10 @@ kubectl logs -l app.kubernetes.io/name=dataflow-operator | grep -i secret
 
 ## Рекомендации по производительности
 
+### Общие настройки spec
+
+- **channelBufferSize** (по умолчанию 100): размер буфера каналов между source, processor и sink. При высокой нагрузке Kafka (десятки тысяч msg/s) увеличьте до 500–1000, чтобы снизить блокировки, когда sink медленнее source.
+
 ### Kafka
 
 - Используйте несколько брокеров для отказоустойчивости
@@ -1313,10 +1317,11 @@ kubectl logs -l app.kubernetes.io/name=dataflow-operator | grep -i secret
 
 ### Проблемы с производительностью
 
-1. Увеличьте размер батчей для приемников
-2. Настройте правильный `pollInterval` для источников
-3. Используйте несколько инстансов оператора для масштабирования
-4. Мониторьте метрики обработки сообщений
+1. Увеличьте `channelBufferSize` (500–1000) при высокой нагрузке Kafka
+2. Увеличьте размер батчей для приемников
+3. Настройте правильный `pollInterval` для источников
+4. Используйте несколько инстансов оператора для масштабирования
+5. Мониторьте метрики обработки сообщений
 
 ### Логирование
 

docs/ru/examples.md

@@ -644,6 +644,28 @@ status:
 - Настройте правильные consumer groups для Kafka
 - Мониторьте статус DataFlow ресурсов
 
+## Высоконагруженный Kafka-пайплайн
+
+При высокой скорости сообщений Kafka (десятки тысяч msg/s) увеличьте `channelBufferSize` и `batchSize` в sink:
+
+```yaml
+spec:
+  channelBufferSize: 500   # по умолчанию 100; снижает блокировки, когда sink медленнее source
+  source:
+    type: kafka
+    config:
+      brokers: [localhost:9092]
+      topic: high-volume-topic
+      consumerGroup: dataflow-group
+  sink:
+    type: postgresql
+    config:
+      connectionString: "..."
+      table: events
+      batchSize: 500
+      batchFlushIntervalSeconds: 2
+```
+
 ## Настройка ресурсов и размещения подов
 
 Каждый ресурс DataFlow создает отдельный под (Deployment) для обработки данных. Вы можете настроить ресурсы, выбор нод, affinity и tolerations для этих подов.