specgen: re-enable unit tests (#217)

benchmark.go

@@ -14,8 +14,6 @@
 
 package sdk
 
-/*
-
 import (
 	"context"
 	"fmt"
@@ -61,7 +59,6 @@ type benchmarkSource struct {
 	config map[string]string
 
 	// measures
-	configure time.Duration
 	open      time.Duration
 	firstRead time.Duration
 	allReads  time.Duration
@@ -75,13 +72,6 @@ func (bm *benchmarkSource) Run(b *testing.B) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	bm.configure = bm.measure(func() {
-		err := bm.source.Configure(ctx, bm.config)
-		if err != nil {
-			b.Fatal(err)
-		}
-	})
-
 	bm.open = bm.measure(func() {
 		err := bm.source.Open(ctx, nil)
 		if err != nil {
@@ -174,7 +164,6 @@ func (*benchmarkSource) measure(f func()) time.Duration {
 func (bm *benchmarkSource) reportMetrics(b *testing.B) {
 	b.ReportMetric(0, "ns/op") // suppress ns/op metric, it is misleading in this benchmarkSource
 
-	b.ReportMetric(bm.configure.Seconds(), "configure")
 	b.ReportMetric(bm.open.Seconds(), "open")
 	b.ReportMetric(bm.stop.Seconds(), "stop")
 	b.ReportMetric(bm.teardown.Seconds(), "teardown")
@@ -185,6 +174,3 @@ func (bm *benchmarkSource) reportMetrics(b *testing.B) {
 	b.ReportMetric(bm.firstAck.Seconds(), "firstAck")
 	b.ReportMetric(float64(b.N-1)/bm.allAcks.Seconds(), "acks/s")
 }
-
-
-*/

destination_middleware.go

@@ -126,7 +126,7 @@ type DestinationWithBatch struct {
 	// Maximum size of batch before it gets written to the destination.
 	BatchSize int `json:"sdk.batch.size" default:"0" validate:"gt=-1"`
 	// Maximum delay before an incomplete batch is written to the destination.
-	BatchDelay time.Duration `json:"sdk.batch.delay" default:"0" validate:"gt=-1"`
+	BatchDelay time.Duration `json:"sdk.batch.delay" default:"0"`
 }
 
 // Wrap a Destination into the middleware.
@@ -164,8 +164,8 @@ func (d *destinationWithBatch) Open(ctx context.Context) error {
 	return nil
 }
 
-// setBatchEnabled stores the boolean in the context. If the context already
-// contains the key it will update the boolean under that key and return the
+// setBatchConfig stores a DestinationWithBatch instance in the context. If the context already
+// contains the key it will update the DestinationWithBatch under that key and return the
 // same context, otherwise it will return a new context with the stored value.
 // This is used to signal to destinationPluginAdapter if the Destination is
 // wrapped into DestinationWithBatchConfig middleware.
@@ -174,7 +174,7 @@ func (*destinationWithBatch) setBatchConfig(ctx context.Context, cfg Destination
 	if ok {
 		*ctxCfg = cfg
 	} else {
-		ctx = context.WithValue(ctx, ctxKeyBatchConfig{}, cfg)
+		ctx = context.WithValue(ctx, ctxKeyBatchConfig{}, &cfg)
 	}
 	return ctx
 }
@@ -425,26 +425,22 @@ type destinationWithSchemaExtraction struct {
 	Destination
 	config *DestinationWithSchemaExtraction
 
-	payloadEnabled bool
-	keyEnabled     bool
-
 	payloadWarnOnce sync.Once
 	keyWarnOnce     sync.Once
 }
 
 func (d *destinationWithSchemaExtraction) Write(ctx context.Context, records []opencdc.Record) (int, error) {
-	if d.keyEnabled {
-		for i := range records {
+	for i := range records {
+		if *d.config.KeyEnabled {
 			if err := d.decodeKey(ctx, &records[i]); err != nil {
 				if len(records) > 0 {
 					err = fmt.Errorf("record %d: %w", i, err)
 				}
 				return 0, err
 			}
 		}
-	}
-	if d.payloadEnabled {
-		for i := range records {
+
+		if *d.config.PayloadEnabled {
 			if err := d.decodePayload(ctx, &records[i]); err != nil {
 				if len(records) > 0 {
 					err = fmt.Errorf("record %d: %w", i, err)