oracledb_cdc: add source_ts_ms metadata (#4250)

Author: josephwoodward

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

@@ -120,6 +120,7 @@ This input adds the following metadata fields to each message:
 - table_name: Name of the table that the message originated from.
 - operation: Type of operation that generated the message: "read", "delete", "insert", or "update". "read" is from messages that are read in the initial snapshot phase.
 - scn: the System Change Number in Oracle.
+- source_ts_ms: The timestamp of when Oracle wrote the change record into the redo log, expressed as milliseconds since the Unix epoch. This reflects the database server's wall-clock time at the moment the DML executed, not the transaction commit time.
 - schema: The table schema, for use with schema-aware downstream processors such as `schema_registry_encode`. When new columns are detected in CDC events, the schema is automatically refreshed from the Oracle catalog. Dropped columns are reflected after a connector restart.
 
 == Permissions

internal/impl/oracledb/batcher.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -166,6 +167,12 @@ func (b *batchPublisher) Publish(ctx context.Context, m *replication.MessageEven
 	if m.SCN.IsValid() {
 		msg.MetaSet("scn", m.SCN.String())
 	}
+	if !m.Timestamp.IsZero() {
+		// upcon connection go-ora automatically queries the server's timezone and stores
+		// it in conn.dbServerTimeZone so it can convert the redo log timestamp
+		// from database-local time to UTC
+		msg.MetaSet("source_ts_ms", strconv.FormatInt(m.Timestamp.UnixMilli(), 10))
+	}
 	if m.CheckpointSCN.IsValid() {
 		msg.MetaSet("checkpoint_scn", m.CheckpointSCN.String())
 	}

internal/impl/oracledb/input_oracledb_cdc.go

@@ -76,6 +76,7 @@ This input adds the following metadata fields to each message:
 - table_name: Name of the table that the message originated from.
 - operation: Type of operation that generated the message: "read", "delete", "insert", or "update". "read" is from messages that are read in the initial snapshot phase.
 - scn: the System Change Number in Oracle.
+- source_ts_ms: The timestamp of when Oracle wrote the change record into the redo log, expressed as milliseconds since the Unix epoch. This reflects the database server's wall-clock time at the moment the DML executed, not the transaction commit time.
 - schema: The table schema, for use with schema-aware downstream processors such as ` + "`schema_registry_encode`" + `. When new columns are detected in CDC events, the schema is automatically refreshed from the Oracle catalog. Dropped columns are reflected after a connector restart.
 
 == Permissions

internal/impl/oracledb/integration_test.go

@@ -13,6 +13,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -541,6 +542,14 @@ oracledb_cdc:
 			op, ok := msg.MetaGet("operation")
 			require.Truef(t, ok, "message %d missing 'operation' metadata", i)
 			assert.Equalf(t, operation, op, "message %d: expected operation '%s', got %q", i, operation, op)
+
+			tsStr, ok := msg.MetaGet("source_ts_ms")
+			require.Truef(t, ok, "message %d missing 'source_ts_ms' metadata", i)
+			tsMs, err := strconv.ParseInt(tsStr, 10, 64)
+			require.NoErrorf(t, err, "message %d: source_ts_ms %q is not a valid int64", i, tsStr)
+			tsTime := time.UnixMilli(tsMs)
+			assert.Truef(t, tsTime.After(time.Now().Add(-5*time.Minute)), "message %d: source_ts_ms %d is too far in the past", i, tsMs)
+			assert.Truef(t, tsTime.Before(time.Now().Add(time.Minute)), "message %d: source_ts_ms %d is in the future", i, tsMs)
 		}
 
 		for _, expectedKey := range []string{"TESTDB.FOO", "TESTDB.FOO2", "TESTDB2.BAR"} {

internal/impl/oracledb/replication/stream_message.go

@@ -117,12 +117,12 @@ type ColumnMeta struct {
 
 // MessageEvent represents a single change from Table's change table in the database.
 type MessageEvent struct {
-	SCN           SCN          `json:"start_scn"`
-	CheckpointSCN SCN          `json:"-"`
-	Operation     OpType       `json:"operation"`
-	Schema        string       `json:"schema"`
-	Table         string       `json:"table"`
-	Data          any          `json:"data"`
-	Timestamp     time.Time    `json:"timestamp"`
-	ColumnMeta    []ColumnMeta `json:"-"`
+	SCN           SCN
+	CheckpointSCN SCN
+	Operation     OpType
+	Schema        string
+	Table         string
+	Data          any
+	Timestamp     time.Time
+	ColumnMeta    []ColumnMeta
 }