feat: add support for collapsing sink batches based on message acknowledgment

- Introduced CollapseBatchOnMessageAck field in DataFlowSpec to control whether message acknowledgment forces MaxBatchSize=1.
- Implemented logic in the processor and connectors to respect the new collapsing behavior.
- Added unit tests to validate the new collapsing functionality for various scenarios.
- Updated CRDs and documentation to reflect the new CollapseBatchOnMessageAck configuration.

Author: ilyario

api/v1/dataflow_ack.go

@@ -35,3 +35,17 @@ func AckGranularityOrDefault(spec *DataFlowSpec) string {
 func AckGranularityIsMessage(spec *DataFlowSpec) bool {
 	return AckGranularityOrDefault(spec) == AckGranularityMessage
 }
+
+// CollapseBatchOnMessageAckOrDefault reports whether message-ack must force MaxBatchSize=1.
+// Default true preserves legacy coupling; false keeps sink batchSize with per-message source commit.
+func CollapseBatchOnMessageAckOrDefault(spec *DataFlowSpec) bool {
+	if spec != nil && spec.CollapseBatchOnMessageAck != nil {
+		return *spec.CollapseBatchOnMessageAck
+	}
+	return true
+}
+
+// ShouldCollapseSinkBatch reports whether sinks must force MaxBatchSize=1 for the given spec.
+func ShouldCollapseSinkBatch(spec *DataFlowSpec) bool {
+	return AckGranularityIsMessage(spec) && CollapseBatchOnMessageAckOrDefault(spec)
+}

api/v1/dataflow_ack_test.go

@@ -38,3 +38,33 @@ func TestAckGranularityIsMessage(t *testing.T) {
 	assert.False(t, AckGranularityIsMessage(&DataFlowSpec{}))
 	assert.True(t, AckGranularityIsMessage(&DataFlowSpec{AckGranularity: AckGranularityMessage}))
 }
+
+func TestCollapseBatchOnMessageAckOrDefault(t *testing.T) {
+	t.Parallel()
+
+	assert.True(t, CollapseBatchOnMessageAckOrDefault(nil))
+	assert.True(t, CollapseBatchOnMessageAckOrDefault(&DataFlowSpec{}))
+
+	f := false
+	assert.False(t, CollapseBatchOnMessageAckOrDefault(&DataFlowSpec{CollapseBatchOnMessageAck: &f}))
+	tr := true
+	assert.True(t, CollapseBatchOnMessageAckOrDefault(&DataFlowSpec{CollapseBatchOnMessageAck: &tr}))
+}
+
+func TestShouldCollapseSinkBatch(t *testing.T) {
+	t.Parallel()
+
+	assert.False(t, ShouldCollapseSinkBatch(nil))
+	assert.False(t, ShouldCollapseSinkBatch(&DataFlowSpec{}))
+
+	f := false
+	assert.True(t, ShouldCollapseSinkBatch(&DataFlowSpec{AckGranularity: AckGranularityMessage}))
+	assert.False(t, ShouldCollapseSinkBatch(&DataFlowSpec{
+		AckGranularity:            AckGranularityMessage,
+		CollapseBatchOnMessageAck: &f,
+	}))
+	assert.False(t, ShouldCollapseSinkBatch(&DataFlowSpec{
+		AckGranularity:            AckGranularityBatch,
+		CollapseBatchOnMessageAck: &f,
+	}))
+}

api/v1/dataflow_types.go

@@ -96,6 +96,15 @@ type DataFlowSpec struct {
 	// +optional
 	AckGranularity string `json:"ackGranularity,omitempty"`
 
+	// CollapseBatchOnMessageAck forces sink MaxBatchSize=1 when ackGranularity is message.
+	// Default: true (legacy coupling of ack semantics and write-batch size).
+	// Set false to keep sink.config.batchSize while still committing source offsets
+	// per message after a successful bulk write (watermark-style message ack).
+	// Ignored when ackGranularity is batch.
+	// +kubebuilder:default:=true
+	// +optional
+	CollapseBatchOnMessageAck *bool `json:"collapseBatchOnMessageAck,omitempty"`
+
 	// CheckpointReset clears persisted source checkpoint on the next processor start (one-shot).
 	// Alternatively set annotation dataflow.dataflow.io/reset-checkpoint: "true" on the DataFlow.
 	// +optional

api/v1/zz_generated.deepcopy.go

@@ -989,6 +989,15 @@ spec:
                   Shrinks the duplicate window on pod crash from the debounce interval to roughly one batch.
                   Default: false (debounced save only). Recommended for migration and cron workloads.
                 type: boolean
+              collapseBatchOnMessageAck:
+                default: true
+                description: |-
+                  CollapseBatchOnMessageAck forces sink MaxBatchSize=1 when ackGranularity is message.
+                  Default: true (legacy coupling of ack semantics and write-batch size).
+                  Set false to keep sink.config.batchSize while still committing source offsets
+                  per message after a successful bulk write (watermark-style message ack).
+                  Ignored when ackGranularity is batch.
+                type: boolean
               concurrencyPolicy:
                 type: string
               errors:

config/crd/bases/dataflow.dataflow.io_dataflowcrons.yaml

@@ -991,6 +991,15 @@ spec:
                   Shrinks the duplicate window on pod crash from the debounce interval to roughly one batch.
                   Default: false (debounced save only). Recommended for migration and cron workloads.
                 type: boolean
+              collapseBatchOnMessageAck:
+                default: true
+                description: |-
+                  CollapseBatchOnMessageAck forces sink MaxBatchSize=1 when ackGranularity is message.
+                  Default: true (legacy coupling of ack semantics and write-batch size).
+                  Set false to keep sink.config.batchSize while still committing source offsets
+                  per message after a successful bulk write (watermark-style message ack).
+                  Ignored when ackGranularity is batch.
+                type: boolean
               errors:
                 description: Errors defines the error sink for messages that failed
                   to be written to the main sink

config/crd/bases/dataflow.dataflow.io_dataflows.yaml

@@ -5,8 +5,9 @@ metadata:
 spec:
   # Высоконагруженный поток: увеличиваем буфер канала
   channelBufferSize: 1000
-  # Для high-volume потоков используем message-level ack
+  # Message-level source commit without collapsing ClickHouse bulk inserts
   ackGranularity: message
+  collapseBatchOnMessageAck: false
   source:
     type: kafka
     config:

config/samples/kafka-to-clickhouse-high-volume.yaml

@@ -173,10 +173,11 @@ func (c *connectorMetadata) SetMetadata(namespace, name string) {
 
 // progressRecorder is an optional callback invoked after successful pipeline progress (e.g. sink flush + ack).
 type progressRecorder struct {
-	onProgress         func()
-	batchAckSyncer     checkpoint.BatchAckSyncer
-	ackGranularity     string
-	checkpointReporter checkpointSaveReporter
+	onProgress                func()
+	batchAckSyncer            checkpoint.BatchAckSyncer
+	ackGranularity            string
+	collapseBatchOnMessageAck *bool // nil → true (legacy) when message-ack
+	checkpointReporter        checkpointSaveReporter
 }
 
 func (p *progressRecorder) setReporter(r checkpointSaveReporter) {
@@ -202,10 +203,27 @@ func (p *progressRecorder) SetAckGranularity(granularity string) {
 	p.ackGranularity = v1.AckGranularityBatch
 }
 
+// SetCollapseBatchOnMessageAck controls whether message-ack forces MaxBatchSize=1.
+// Wire from CRD via processor (default true when unset on the spec).
+func (p *progressRecorder) SetCollapseBatchOnMessageAck(collapse bool) {
+	p.collapseBatchOnMessageAck = &collapse
+}
+
 func (p *progressRecorder) ackGranularityIsMessage() bool {
 	return p.ackGranularity == v1.AckGranularityMessage
 }
 
+// shouldCollapseBatchForAck reports whether sink batching must be forced to 1.
+func (p *progressRecorder) shouldCollapseBatchForAck() bool {
+	if !p.ackGranularityIsMessage() {
+		return false
+	}
+	if p.collapseBatchOnMessageAck != nil {
+		return *p.collapseBatchOnMessageAck
+	}
+	return true
+}
+
 func (p *progressRecorder) notifyProgress() {
 	if p.onProgress != nil {
 		p.onProgress()

internal/connectors/base.go

@@ -150,3 +150,22 @@ func TestAckAfterSuccessfulWrite_MessageGranularity(t *testing.T) {
 	assert.Equal(t, int32(2), progressCalls.Load())
 	assert.Equal(t, int32(2), syncer.calls.Load())
 }
+
+func TestShouldCollapseBatchForAck(t *testing.T) {
+	t.Parallel()
+
+	rec := &progressRecorder{}
+	assert.False(t, rec.shouldCollapseBatchForAck())
+
+	rec.SetAckGranularity("message")
+	assert.True(t, rec.shouldCollapseBatchForAck(), "default collapses when message-ack")
+
+	rec.SetCollapseBatchOnMessageAck(false)
+	assert.False(t, rec.shouldCollapseBatchForAck())
+
+	rec.SetCollapseBatchOnMessageAck(true)
+	assert.True(t, rec.shouldCollapseBatchForAck())
+
+	rec.SetAckGranularity("batch")
+	assert.False(t, rec.shouldCollapseBatchForAck(), "batch ack never collapses via this helper")
+}

internal/connectors/base_progress_test.go

@@ -65,9 +65,10 @@ func NewBatchWriteConfig(batchSize, flushIntervalSec *int32, defaultBatchSize in
 	}
 }
 
-// ApplyAckGranularity forces single-message batches when per-message ack is enabled.
-func ApplyAckGranularity(cfg BatchWriteConfig, messageAck bool) BatchWriteConfig {
-	if messageAck {
+// ApplyAckGranularity forces single-message batches when collapseBatch is true
+// (typically message-ack with collapseBatchOnMessageAck enabled / default).
+func ApplyAckGranularity(cfg BatchWriteConfig, collapseBatch bool) BatchWriteConfig {
+	if collapseBatch {
 		cfg.MaxBatchSize = 1
 	}
 	return cfg