oracledb_cdc: improve error handling

Author: josephwoodward

internal/impl/oracledb/logminer/logminer.go

@@ -288,7 +288,9 @@ func (lm *LogMiner) processRedoEvent(ctx context.Context, redoEvent *sqlredo.Red
 		}
 
 		if lm.cfg.LOBEnabled && (redoEvent.Operation == sqlredo.OpInsert || redoEvent.Operation == sqlredo.OpUpdate) {
-			lm.drainPendingLOBWrites(ctx, redoEvent.TransactionID)
+			if err = lm.drainPendingLOBWrites(ctx, redoEvent.TransactionID); err != nil {
+				return fmt.Errorf("draining pending lob writes: %w", err)
+			}
 		}
 
 	case sqlredo.OpSelectLobLocator:
@@ -420,8 +422,7 @@ func (lm *LogMiner) processRedoEvent(ctx context.Context, redoEvent *sqlredo.Red
 		// not as HEXTORAW. Only BLOB uses binary/hex encoding.
 		writeInfo, err := sqlredo.ParseLobWrite(redoEvent.SQLRedo.String, acc.IsBinary)
 		if err != nil {
-			lm.log.Warnf("Failed to parse LOB_WRITE SQL (scn=%d, txn=%s): %v\nSQL: %.500s", redoEvent.SCN, redoEvent.TransactionID, err, redoEvent.SQLRedo.String)
-			return nil
+			return fmt.Errorf("parsing LOB_WRITE SQL (scn=%d, txn=%s): %v\nSQL: %.500s", redoEvent.SCN, redoEvent.TransactionID, err, redoEvent.SQLRedo.String)
 		}
 		acc.AddFragment(writeInfo.Offset, writeInfo.Data)
 
@@ -766,10 +767,10 @@ func (lm *LogMiner) inferLOBLocator(ctx context.Context, event *sqlredo.RedoEven
 // drainPendingLOBWrites retries LOB_WRITE events that were buffered because their
 // INSERT had not yet been processed. Called after each INSERT or UPDATE is added
 // to the transaction cache, at which point inferLOBLocator should succeed.
-func (lm *LogMiner) drainPendingLOBWrites(ctx context.Context, txnID sqlredo.TransactionID) {
+func (lm *LogMiner) drainPendingLOBWrites(ctx context.Context, txnID sqlredo.TransactionID) error {
 	pending := lm.pendingLOBWrites[txnID]
 	if len(pending) == 0 {
-		return
+		return nil
 	}
 
 	var remaining []*sqlredo.RedoEvent
@@ -792,8 +793,7 @@ func (lm *LogMiner) drainPendingLOBWrites(ctx context.Context, txnID sqlredo.Tra
 		}
 		writeInfo, err := sqlredo.ParseLobWrite(event.SQLRedo.String, acc.IsBinary)
 		if err != nil {
-			lm.log.Warnf("Failed to parse buffered LOB_WRITE SQL (scn=%d, txn=%s): %v\nSQL: %.500s", event.SCN, txnID, err, event.SQLRedo.String)
-			continue
+			return fmt.Errorf("parsing buffered LOB_WRITE SQL (scn=%d, txn=%s): %v\nSQL: %.500s", event.SCN, txnID, err, event.SQLRedo.String)
 		}
 		acc.AddFragment(writeInfo.Offset, writeInfo.Data)
 		lm.log.Debugf("Drained buffered LOB_WRITE for %s.%s.%s (txn=%s)", acc.Schema, acc.Table, acc.Column, txnID)
@@ -804,6 +804,8 @@ func (lm *LogMiner) drainPendingLOBWrites(ctx context.Context, txnID sqlredo.Tra
 	} else {
 		lm.pendingLOBWrites[txnID] = remaining
 	}
+
+	return nil
 }
 
 func (lm *LogMiner) queryLogMinerContents(ctx context.Context, conn *sql.Conn, startSCN, endSCN uint64, processEvent func(context.Context, *sqlredo.RedoEvent) error) error {