Enhance DataFlowSpec with new features: replace Errors field type with ErrorSinkSpec, add CheckpointReset and StrictIdempotency fields. Update validation and processing logic to accommodate these changes, including new tests for error sink configuration and checkpoint handling.

Author: ilyario

api/v1/dataflow_checkpoint_reset.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+// AnnotationResetCheckpoint requests a one-shot checkpoint reset on the next processor run.
+const AnnotationResetCheckpoint = "dataflow.dataflow.io/reset-checkpoint"
+
+// CheckpointResetRequested reports whether the processor should clear persisted checkpoint on startup.
+func CheckpointResetRequested(spec *DataFlowSpec) bool {
+	return spec != nil && spec.CheckpointReset != nil && *spec.CheckpointReset
+}
+
+// StrictIdempotencyEnabled reports whether non-idempotent polling pipelines are rejected at admission.
+func StrictIdempotencyEnabled(spec *DataFlowSpec) bool {
+	return spec != nil && spec.StrictIdempotency != nil && *spec.StrictIdempotency
+}

api/v1/dataflow_errors.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+const (
+	// ErrorAckPolicyAfterWrite commits source offsets after the error sink successfully writes (default).
+	ErrorAckPolicyAfterWrite = "afterWrite"
+	// ErrorAckPolicyNever does not commit source offsets for messages sent to the error sink.
+	ErrorAckPolicyNever = "never"
+	// ErrorAckPolicyAfterMainSinkSuccess commits offsets only after main sink success (same as never on error path).
+	ErrorAckPolicyAfterMainSinkSuccess = "afterMainSinkSuccess"
+)
+
+// ErrorSinkSpec defines the error sink and how source ack behaves for failed messages.
+type ErrorSinkSpec struct {
+	SinkSpec `json:",inline"`
+
+	// AckPolicy controls when source offsets/checkpoints are committed for messages routed to the error sink.
+	// afterWrite (default): ack after error sink write.
+	// never: do not ack failed messages (they may be re-read on restart).
+	// afterMainSinkSuccess: ack only after main sink success (failed messages are not acked).
+	// +kubebuilder:validation:Enum=afterWrite;never;afterMainSinkSuccess
+	// +kubebuilder:default:=afterWrite
+	// +optional
+	AckPolicy string `json:"ackPolicy,omitempty"`
+}
+
+// ErrorAckPolicyOrDefault returns the configured error sink ack policy.
+func ErrorAckPolicyOrDefault(errors *ErrorSinkSpec) string {
+	if errors == nil || errors.AckPolicy == "" {
+		return ErrorAckPolicyAfterWrite
+	}
+	switch errors.AckPolicy {
+	case ErrorAckPolicyNever, ErrorAckPolicyAfterMainSinkSuccess:
+		return errors.AckPolicy
+	default:
+		return ErrorAckPolicyAfterWrite
+	}
+}
+
+// ShouldAckOnErrorSink reports whether source ack should propagate when writing to the error sink.
+func ShouldAckOnErrorSink(errors *ErrorSinkSpec) bool {
+	switch ErrorAckPolicyOrDefault(errors) {
+	case ErrorAckPolicyNever, ErrorAckPolicyAfterMainSinkSuccess:
+		return false
+	default:
+		return true
+	}
+}

api/v1/dataflow_errors_test.go

@@ -0,0 +1,42 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestShouldAckOnErrorSink(t *testing.T) {
+	t.Parallel()
+
+	assert.True(t, ShouldAckOnErrorSink(nil))
+	assert.True(t, ShouldAckOnErrorSink(&ErrorSinkSpec{}))
+	assert.True(t, ShouldAckOnErrorSink(&ErrorSinkSpec{AckPolicy: ErrorAckPolicyAfterWrite}))
+	assert.False(t, ShouldAckOnErrorSink(&ErrorSinkSpec{AckPolicy: ErrorAckPolicyNever}))
+	assert.False(t, ShouldAckOnErrorSink(&ErrorSinkSpec{AckPolicy: ErrorAckPolicyAfterMainSinkSuccess}))
+}
+
+func TestCheckpointResetRequested(t *testing.T) {
+	t.Parallel()
+
+	assert.False(t, CheckpointResetRequested(nil))
+	assert.False(t, CheckpointResetRequested(&DataFlowSpec{}))
+	trueVal := true
+	assert.True(t, CheckpointResetRequested(&DataFlowSpec{CheckpointReset: &trueVal}))
+}

api/v1/dataflow_idempotency.go

@@ -0,0 +1,105 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"github.com/dataflow-operator/dataflow/pkg/providers"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+)
+
+func validateErrors(errors *ErrorSinkSpec, f *field.Path) field.ErrorList {
+	var all field.ErrorList
+	if errors == nil {
+		return all
+	}
+	all = append(all, validateSink(&errors.SinkSpec, f)...)
+	if errors.AckPolicy != "" &&
+		errors.AckPolicy != ErrorAckPolicyAfterWrite &&
+		errors.AckPolicy != ErrorAckPolicyNever &&
+		errors.AckPolicy != ErrorAckPolicyAfterMainSinkSuccess {
+		all = append(all, field.NotSupported(f.Child("ackPolicy"), errors.AckPolicy, []string{
+			ErrorAckPolicyAfterWrite,
+			ErrorAckPolicyNever,
+			ErrorAckPolicyAfterMainSinkSuccess,
+		}))
+	}
+	return all
+}
+
+func validateIdempotency(spec *DataFlowSpec, f *field.Path) field.ErrorList {
+	var all field.ErrorList
+	if spec == nil || !isPollingSourceType(spec.Source.Type) {
+		return all
+	}
+	if sinkIsIdempotent(&spec.Sink) {
+		return all
+	}
+	msg := "polling source with non-idempotent main sink may produce duplicates on restart; enable upsertMode on the sink or set strictIdempotency: false to accept this warning"
+	if StrictIdempotencyEnabled(spec) {
+		all = append(all, field.Invalid(f.Child("sink"), spec.Sink.Type,
+			"strictIdempotency is enabled but main sink is not idempotent (enable upsertMode for postgresql/trino/clickhouse sinks)"))
+	} else {
+		_ = msg // warnings emitted via WarnDataFlowSpec
+	}
+	return all
+}
+
+// WarnDataFlowSpec returns admission warnings for non-fatal spec issues.
+func WarnDataFlowSpec(spec *DataFlowSpec) admission.Warnings {
+	var warnings admission.Warnings
+	if spec == nil {
+		return warnings
+	}
+	if isPollingSourceType(spec.Source.Type) && !sinkIsIdempotent(&spec.Sink) && !StrictIdempotencyEnabled(spec) {
+		warnings = append(warnings,
+			"polling source with non-idempotent main sink may produce duplicates on restart; enable sink.config.upsertMode or set strictIdempotency: true to reject at admission")
+	}
+	return warnings
+}
+
+func isPollingSourceType(sourceType string) bool {
+	return providers.SourceSupportsCheckpoint(sourceType)
+}
+
+func sinkIsIdempotent(sink *SinkSpec) bool {
+	if sink == nil || sink.Config == nil || len(sink.Config.Raw) == 0 {
+		return false
+	}
+	switch sink.Type {
+	case "postgresql":
+		cfg, err := sink.GetPostgreSQLConfig()
+		if err != nil || cfg == nil {
+			return false
+		}
+		return cfg.UpsertMode != nil && *cfg.UpsertMode
+	case "clickhouse":
+		cfg, err := sink.GetClickHouseConfig()
+		if err != nil || cfg == nil {
+			return false
+		}
+		return cfg.UpsertMode != nil && *cfg.UpsertMode
+	case "trino":
+		cfg, err := sink.GetTrinoConfig()
+		if err != nil || cfg == nil {
+			return false
+		}
+		return cfg.UpsertMode != nil && *cfg.UpsertMode
+	default:
+		return false
+	}
+}

api/v1/dataflow_idempotency_test.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWarnDataFlowSpec_pollingNonIdempotent(t *testing.T) {
+	t.Parallel()
+
+	upsertOff := false
+	spec := DataFlowSpec{
+		Source: SourceSpec{Type: "postgresql", Config: mustConfig(PostgreSQLSourceSpec{ConnectionString: "postgres://u:p@localhost/db", Table: "t"})},
+		Sink: SinkSpec{
+			Type:   "postgresql",
+			Config: mustConfig(PostgreSQLSinkSpec{ConnectionString: "postgres://u:p@localhost/db", Table: "out", UpsertMode: &upsertOff}),
+		},
+	}
+	warnings := WarnDataFlowSpec(&spec)
+	assert.Len(t, warnings, 1)
+}
+
+func TestValidateDataFlowSpec_strictIdempotency(t *testing.T) {
+	t.Parallel()
+
+	strict := true
+	spec := DataFlowSpec{
+		StrictIdempotency: &strict,
+		Source:            SourceSpec{Type: "postgresql", Config: mustConfig(PostgreSQLSourceSpec{ConnectionString: "postgres://u:p@localhost/db", Table: "t"})},
+		Sink: SinkSpec{
+			Type:   "postgresql",
+			Config: mustConfig(PostgreSQLSinkSpec{ConnectionString: "postgres://u:p@localhost/db", Table: "out"}),
+		},
+	}
+	errs := ValidateDataFlowSpec(&spec)
+	assert.NotEmpty(t, errs)
+}
+
+func TestValidateErrors_ackPolicy(t *testing.T) {
+	t.Parallel()
+
+	spec := DataFlowSpec{
+		Source: SourceSpec{Type: "kafka", Config: mustConfig(KafkaSourceSpec{Brokers: []string{"b"}, Topic: "t"})},
+		Sink:   SinkSpec{Type: "kafka", Config: mustConfig(KafkaSinkSpec{Brokers: []string{"b"}, Topic: "out"})},
+		Errors: &ErrorSinkSpec{
+			SinkSpec:  SinkSpec{Type: "kafka", Config: mustConfig(KafkaSinkSpec{Brokers: []string{"b"}, Topic: "err"})},
+			AckPolicy: "invalid",
+		},
+	}
+	errs := ValidateDataFlowSpec(&spec)
+	assert.NotEmpty(t, errs)
+}

api/v1/dataflow_types.go

@@ -38,7 +38,7 @@ type DataFlowSpec struct {
 
 	// Errors defines the error sink for messages that failed to be written to the main sink
 	// +optional
-	Errors *SinkSpec `json:"errors,omitempty"`
+	Errors *ErrorSinkSpec `json:"errors,omitempty"`
 
 	// Resources defines the resource requirements for the processor pod
 	// +optional
@@ -96,6 +96,16 @@ type DataFlowSpec struct {
 	// +optional
 	AckGranularity string `json:"ackGranularity,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
+	CheckpointReset *bool `json:"checkpointReset,omitempty"`
+
+	// StrictIdempotency rejects polling sources paired with non-idempotent main sinks at admission.
+	// When false (default), a warning is emitted instead.
+	// +optional
+	StrictIdempotency *bool `json:"strictIdempotency,omitempty"`
+
 	// ChannelBufferSize is the buffer size for message channels between source, processor, and sink.
 	// Larger values reduce blocking when sink is slower than source (e.g. high Kafka throughput).
 	// Default: 100. Recommended for high throughput: 500–1000.

api/v1/dataflow_validation.go

@@ -40,12 +40,13 @@ func ValidateDataFlowSpec(spec *DataFlowSpec) field.ErrorList {
 	all = append(all, validateSource(&spec.Source, f.Child("source"))...)
 	all = append(all, validateSink(&spec.Sink, f.Child("sink"))...)
 	if spec.Errors != nil {
-		all = append(all, validateSink(spec.Errors, f.Child("errors"))...)
+		all = append(all, validateErrors(spec.Errors, f.Child("errors"))...)
 	}
 	all = append(all, validateTransformations(spec.Transformations, f.Child("transformations"))...)
 	all = append(all, validateResources(spec.Resources, f.Child("resources"))...)
 	all = append(all, validateReplicas(spec, f)...)
 	all = append(all, validateAckGranularity(spec, f)...)
+	all = append(all, validateIdempotency(spec, f)...)
 	return all
 }
 

api/v1/dataflow_webhook.go

@@ -30,10 +30,10 @@ func (r *DataFlow) ValidateCreate(ctx context.Context, obj runtime.Object) (admi
 		return nil, nil
 	}
 	errs := ValidateDataFlowSpec(&df.Spec)
-	if len(errs) == 0 {
-		return nil, nil
+	if len(errs) > 0 {
+		return nil, errs.ToAggregate()
 	}
-	return nil, errs.ToAggregate()
+	return WarnDataFlowSpec(&df.Spec), nil
 }
 
 // ValidateUpdate implements admission.CustomValidator.
@@ -43,10 +43,10 @@ func (r *DataFlow) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 		return nil, nil
 	}
 	errs := ValidateDataFlowSpec(&df.Spec)
-	if len(errs) == 0 {
-		return nil, nil
+	if len(errs) > 0 {
+		return nil, errs.ToAggregate()
 	}
-	return nil, errs.ToAggregate()
+	return WarnDataFlowSpec(&df.Spec), nil
 }
 
 // ValidateDelete implements admission.CustomValidator.