oracledb_cdc: Add support for snapshot_only mode, deprecating stream_snapshot config (#4570)

Author: josephwoodward

.gitignore

@@ -17,3 +17,4 @@ __pycache__
 *.test.exe
 compile_out.txt
 test_output.txt
+benchmark_results.json

docs/modules/components/pages/inputs/oracledb_cdc.adoc

@@ -42,7 +42,7 @@ input:
     connection_string: oracle://username:password@host:port/service_name # No default (required)
     wallet_path: /opt/oracle/wallet # No default (optional)
     wallet_password: "" # No default (optional)
-    stream_snapshot: false
+    snapshot_mode: "" # No default (optional)
     max_parallel_snapshot_tables: 1
     snapshot_max_batch_size: 1000
     logminer:
@@ -84,7 +84,7 @@ input:
     connection_string: oracle://username:password@host:port/service_name # No default (required)
     wallet_path: /opt/oracle/wallet # No default (optional)
     wallet_password: "" # No default (optional)
-    stream_snapshot: false
+    snapshot_mode: "" # No default (optional)
     max_parallel_snapshot_tables: 1
     snapshot_max_batch_size: 1000
     logminer:
@@ -183,20 +183,20 @@ This field contains sensitive information that usually shouldn't be added to a c
 *Type*: `string`
 
 
-=== `stream_snapshot`
+=== `snapshot_mode`
 
-If set to true, the connector will query all the existing data as a part of snapshot process. Otherwise, it will start from the current System Change Number position.
+Controls snapshot behaviour. `none` (default) skips snapshotting and starts streaming from the current SCN. `snapshot_only` performs a full snapshot, persists the SCN checkpoint, then stops without streaming. `snapshot_and_stream` performs a full snapshot then transitions to streaming.
 
 
-*Type*: `bool`
-
-*Default*: `false`
+*Type*: `string`
 
-```yml
-# Examples
+Requires version 4.99.0 or newer
 
-stream_snapshot: true
-```
+Options:
+`none`
+, `snapshot_only`
+, `snapshot_and_stream`
+.
 
 === `max_parallel_snapshot_tables`
 

internal/impl/oracledb/batcher.go

@@ -54,12 +54,7 @@ func newBatchPublisher(batcher *service.Batcher, checkpoint *checkpoint.Capped[r
 // loop creates a long-running process that periodically flushes batches by configured interval.
 // lifted from internal/impl/kafka/franz_reader_ordered.go
 func (p *batchPublisher) loop() {
-	defer func() {
-		if p.batcher != nil {
-			p.batcher.Close(context.Background())
-		}
-		p.shutSig.TriggerHasStopped()
-	}()
+	defer p.shutSig.TriggerHasStopped()
 
 	// No need to loop when there's no batcher for async writes.
 	if p.batcher == nil {
@@ -90,7 +85,10 @@ func (p *batchPublisher) loop() {
 		flushBatch = flushBatchTicker.C
 	}
 
-	closeAtLeisureCtx, done := p.shutSig.SoftStopCtx(context.Background())
+	// hardStopCtx survives a soft stop so that an in-flight publishBatch send can
+	// complete before the loop exits. Only a hard stop (triggered by Close)
+	// cancels it, which is the forced-shutdown last resort.
+	hardStopCtx, done := p.shutSig.HardStopCtx(context.Background())
 	defer done()
 
 	for {
@@ -112,13 +110,13 @@ func (p *batchPublisher) loop() {
 					return
 				}
 
-				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
+				if sendBatch, _ = p.batcher.Flush(hardStopCtx); len(sendBatch) == 0 {
 					return
 				}
 			}()
 
 			if len(sendBatch) > 0 {
-				if err := p.publishBatch(closeAtLeisureCtx, sendBatch); err != nil {
+				if err := p.publishBatch(hardStopCtx, sendBatch); err != nil {
 					return
 				}
 			}
@@ -275,8 +273,32 @@ func (b *batchPublisher) msgs() <-chan asyncMessage {
 	return b.msgChan
 }
 
+// FlushRemaining stops the loop goroutine and then flushes any partial batch
+// still held in the batcher, blocking until it is consumed by ReadBatch.
+func (b *batchPublisher) FlushRemaining(ctx context.Context) error {
+	if b.batcher == nil {
+		return nil
+	}
+	b.shutSig.TriggerSoftStop()
+	<-b.shutSig.HasStoppedChan()
+
+	b.batcherMu.Lock()
+	remaining, err := b.batcher.Flush(ctx)
+	b.batcherMu.Unlock()
+	if err != nil || len(remaining) == 0 {
+		return err
+	}
+	return b.publishBatch(ctx, remaining)
+}
+
 // Close signals the publisher's loop goroutine to stop and waits for it to exit.
+// TriggerHardStop cancels the HardStopCtx used by publishBatch, unblocking any
+// send that is waiting on msgChan when no consumer is left.
 func (b *batchPublisher) Close() {
 	b.shutSig.TriggerSoftStop()
+	b.shutSig.TriggerHardStop()
 	<-b.shutSig.HasStoppedChan()
+	if b.batcher != nil {
+		_ = b.batcher.Close(context.Background())
+	}
 }

internal/impl/oracledb/bench/benchmark_config.yaml

@@ -6,10 +6,9 @@ input:
     # connection_string: oracle://testdb:testdb123@localhost:1521/TESTPDB
     connection_string: oracle://c%23%23testdb:testdb123@localhost:1521/FREE
     pdb_name: TESTPDB
-    stream_snapshot: false
+    snapshot_mode: snapshot_only
     snapshot_max_batch_size: 160000
     logminer:
-      scn_window_size: 190000
       backoff_interval: 2s
       mining_interval: 0s
       # transaction_cache: testcache
@@ -19,7 +18,6 @@ input:
       - TESTDB.PRODUCTS
       - TESTDB.CART
     batching:
-      count: 140000
       period: 1s
 
 output:

internal/impl/oracledb/config.go

@@ -102,3 +102,42 @@ func parseWalletConfig(conf *service.ParsedConfig, overrides map[string]string)
 
 	return nil
 }
+
+// SnapshotMode controls whether and how an initial table snapshot is taken before streaming begins.
+type SnapshotMode string
+
+const (
+	// SnapshotModeNone skips snapshotting and starts streaming from the current SCN.
+	SnapshotModeNone SnapshotMode = "none"
+	// SnapshotModeSnapshotOnly performs a full snapshot, persists the SCN checkpoint, then stops without streaming.
+	SnapshotModeSnapshotOnly SnapshotMode = "snapshot_only"
+	// SnapshotModeSnapshotAndStream performs a full snapshot then transitions to streaming.
+	SnapshotModeSnapshotAndStream SnapshotMode = "snapshot_and_stream"
+)
+
+// IsSnapshotOnly returns true if the snapshot_mode config is snapshot_only, otherwise false.
+func (s SnapshotMode) IsSnapshotOnly() bool {
+	return s == SnapshotModeSnapshotOnly
+}
+
+// IsSnapshotNone returns true if the snapshot_mode config is none, otherwise false.
+func (s SnapshotMode) IsSnapshotNone() bool {
+	return s == SnapshotModeNone
+}
+
+func parseSnapshotMode(conf *service.ParsedConfig) (SnapshotMode, error) {
+	if conf.Contains(ociFieldSnapshotMode) {
+		if raw, err := conf.FieldString(ociFieldSnapshotMode); err != nil {
+			return SnapshotModeNone, err
+		} else {
+			return SnapshotMode(raw), nil
+		}
+	}
+	// snapshot_mode not set — apply backward compat
+	if streamSnapshot, err := conf.FieldBool(ociFieldStreamSnapshot); err != nil {
+		return SnapshotModeNone, err
+	} else if streamSnapshot {
+		return SnapshotModeSnapshotAndStream, nil
+	}
+	return SnapshotModeNone, nil
+}

internal/impl/oracledb/config_test.go

@@ -164,3 +164,65 @@ func TestBuildConnectionURL(t *testing.T) {
 		})
 	}
 }
+
+func TestParseSnapshotMode(t *testing.T) {
+	const minimalOracleCDCYAML = `connection_string: oracle://user:pass@host:1521/svc
+include:
+  - SCHEMA.TABLE
+logminer: {}
+`
+	tests := []struct {
+		name string
+		yaml string
+		want SnapshotMode
+	}{
+		{
+			name: "omitted defaults to none",
+			yaml: minimalOracleCDCYAML,
+			want: SnapshotModeNone,
+		},
+		{
+			name: "explicit none",
+			yaml: minimalOracleCDCYAML + "snapshot_mode: none\n",
+			want: SnapshotModeNone,
+		},
+		{
+			name: "snapshot_only",
+			yaml: minimalOracleCDCYAML + "snapshot_mode: snapshot_only\n",
+			want: SnapshotModeSnapshotOnly,
+		},
+		{
+			name: "snapshot_and_stream",
+			yaml: minimalOracleCDCYAML + "snapshot_mode: snapshot_and_stream\n",
+			want: SnapshotModeSnapshotAndStream,
+		},
+		{
+			// backward compat: stream_snapshot: true with no snapshot_mode set
+			name: "stream_snapshot true upgrades to snapshot_and_stream",
+			yaml: minimalOracleCDCYAML + "stream_snapshot: true\n",
+			want: SnapshotModeSnapshotAndStream,
+		},
+		{
+			// explicit snapshot_mode: none must win over stream_snapshot: true
+			name: "explicit snapshot_mode none overrides stream_snapshot true",
+			yaml: minimalOracleCDCYAML + "snapshot_mode: none\nstream_snapshot: true\n",
+			want: SnapshotModeNone,
+		},
+		{
+			// explicit snapshot_mode wins over stream_snapshot: true
+			name: "explicit snapshot_mode snapshot_only overrides stream_snapshot true",
+			yaml: minimalOracleCDCYAML + "snapshot_mode: snapshot_only\nstream_snapshot: true\n",
+			want: SnapshotModeSnapshotOnly,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conf, err := oracleDBStreamConfigSpec.ParseYAML(tt.yaml, nil)
+			require.NoError(t, err)
+			got, err := parseSnapshotMode(conf)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

internal/impl/oracledb/input_oracledb_cdc.go

@@ -16,6 +16,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -45,6 +46,7 @@ const (
 	ociFieldCheckpointCacheTableName  = "checkpoint_cache_table_name"
 	ociFieldBatching                  = "batching"
 	ociFieldPDBName                   = "pdb_name"
+	ociFieldSnapshotMode              = "snapshot_mode"
 
 	shutdownTimeout = 5 * time.Second
 
@@ -108,7 +110,16 @@ When using the default Oracle based cache, the Connect user requires permission
 	Field(service.NewBoolField(ociFieldStreamSnapshot).
 		Description("If set to true, the connector will query all the existing data as a part of snapshot process. Otherwise, it will start from the current System Change Number position.").
 		Example(true).
-		Default(false),
+		Default(false).
+		Deprecated(),
+	).
+	Field(service.NewStringEnumField(ociFieldSnapshotMode,
+		string(SnapshotModeNone),
+		string(SnapshotModeSnapshotOnly),
+		string(SnapshotModeSnapshotAndStream)).
+		Description("Controls snapshot behaviour. `none` (default) skips snapshotting and starts streaming from the current SCN. `snapshot_only` performs a full snapshot, persists the SCN checkpoint, then stops without streaming. `snapshot_and_stream` performs a full snapshot then transitions to streaming.").
+		Optional().
+		Version("4.99.0"),
 	).
 	Field(service.NewIntField(ociFieldMaxParallelSnapshotTables).
 		Description("Specifies a number of tables that will be processed in parallel during the snapshot processing stage.").
@@ -202,7 +213,7 @@ type asyncMessage struct {
 // Config is the configuration for a Oracle connector.
 type Config struct {
 	ConnectionString     string
-	StreamSnapshot       bool
+	SnapshotMode         SnapshotMode
 	SnapshotMaxBatchSize int
 	SnapshotMaxWorkers   int
 	TablesFilter         *confx.RegexpFilter
@@ -221,15 +232,16 @@ type oracleDBCDCInput struct {
 	publisher *batchPublisher
 	metrics   *service.Metrics
 
-	stopSig *shutdown.Signaller
-	log     *service.Logger
-	cpCache service.Cache
+	stopSig          *shutdown.Signaller
+	snapshotOnlyDone atomic.Bool
+	log              *service.Logger
+	cpCache          service.Cache
 }
 
 func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {
 	var (
 		connectionString     string
-		streamSnapshot       bool
+		snapshotMode         SnapshotMode
 		snapshotMaxWorkers   int
 		snapshotMaxBatchSize int
 
@@ -250,7 +262,7 @@ func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resource
 	if connectionString, err = conf.FieldString(ociFieldConnectionString); err != nil {
 		return nil, err
 	}
-	if streamSnapshot, err = conf.FieldBool(ociFieldStreamSnapshot); err != nil {
+	if snapshotMode, err = parseSnapshotMode(conf); err != nil {
 		return nil, err
 	}
 	if snapshotMaxWorkers, err = conf.FieldInt(ociFieldMaxParallelSnapshotTables); err != nil {
@@ -335,7 +347,7 @@ func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resource
 	o := oracleDBCDCInput{
 		cfg: Config{
 			ConnectionString:     connectionString,
-			StreamSnapshot:       streamSnapshot,
+			SnapshotMode:         snapshotMode,
 			SnapshotMaxWorkers:   snapshotMaxWorkers,
 			SnapshotMaxBatchSize: snapshotMaxBatchSize,
 			SCNCache:             scnCache,
@@ -501,7 +513,7 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 	)
 
 	// no cached SCN means we're not recovering from a restart
-	if o.cfg.StreamSnapshot && cachedSCN == replication.InvalidSCN {
+	if !o.cfg.SnapshotMode.IsSnapshotNone() && cachedSCN == replication.InvalidSCN {
 		if snapshotter, err = replication.NewSnapshot(ctx, o.cfg.ConnectionString, userTables, o.publisher, o.lmCfg.LOBEnabled, pdbNameForCache, o.log, o.metrics); err != nil {
 			return fmt.Errorf("creating database snapshotter: %w", err)
 		}
@@ -531,8 +543,9 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 		o.log.Infof("No cached SCN found, fetched current position from database: %d", cachedSCN)
 	}
 
-	// Reset our stop signal
+	// Reset our stop signal and snapshot-only completion flag
 	o.stopSig = shutdown.NewSignaller()
+	o.snapshotOnlyDone.Store(false)
 
 	go func() {
 		var (
@@ -563,6 +576,16 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 			o.log.Infof("Successfully captured SCN following snapshot: %d", startSCN)
 		}
 
+		if o.cfg.SnapshotMode.IsSnapshotOnly() {
+			if err = o.publisher.FlushRemaining(softCtx); err != nil {
+				o.log.Errorf("Failed to flush remaining snapshot events: %s", err)
+			}
+			o.log.Infof("Snapshot-only mode complete, stopping at SCN %s", startSCN)
+			o.snapshotOnlyDone.Store(true)
+			o.stopSig.TriggerHasStopped()
+			return
+		}
+
 		// streaming
 		wg, _ := errgroup.WithContext(softCtx)
 		wg.Go(func() error {
@@ -644,6 +667,9 @@ func (o *oracleDBCDCInput) ReadBatch(ctx context.Context) (service.MessageBatch,
 	case m := <-o.publisher.msgs():
 		return m.msg, m.ackFn, nil
 	case <-o.stopSig.HasStoppedChan():
+		if o.snapshotOnlyDone.Load() {
+			return nil, nil, service.ErrEndOfInput
+		}
 		return nil, nil, service.ErrNotConnected
 	case <-ctx.Done():
 		return nil, nil, ctx.Err()