fix: address CodeRabbit review findings on PR #13

- fuzz-test.sh: capture validator exit code without set -e abort
- validator.py: per-node watermark cursors instead of global cursor
- validator.py: remove undefined lm_max/olr_max completion guard
- validator.py: compare before-images for UPDATE/DELETE events
- compare-debezium.py: make FUZZ_LOB exclusion a --exclude-tables CLI flag
- docker-compose-fuzz.yaml: build consumer image with pinned deps
- fuzz-test.sh: propagate RAC workload failures instead of || true
- kafka-consumer.py: periodic trimming of per-event seq maps
- fuzz-workload.sql: per-table ID tracking for UPDATE/DELETE targeting
- fuzz-workload.sql: add UPDATE/DELETE for WIDE, PART, MAXSTR, INTERVAL
- FUZZ-TEST.md: clarify down deletes fuzz-data volume
- FUZZ-TEST-PLAN.md: update topics, status, watermark description

Author: rophy

tests/dbz-twin/compare-debezium.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """Compare Debezium LogMiner vs OLR adapter outputs.
 
-Usage: compare-debezium.py <logminer.jsonl> <olr.jsonl>
+Usage: compare-debezium.py [--exclude-tables T1,T2,...] <logminer.jsonl> <olr.jsonl>
 
 Both inputs are JSONL files with Debezium envelope events:
   {"before":..., "after":..., "source":..., "op":..., "ts_ms":...}
@@ -13,6 +13,7 @@
 Exits 0 on match, 1 on mismatch with diff report.
 """
 
+import argparse
 import json
 import sys
 from collections import defaultdict
@@ -26,9 +27,9 @@
     'X19kZWJleml1bV91bmF2YWlsYWJsZV92YWx1ZQ==',
 }
 
-# Tables to exclude from comparison (stats/bookkeeping, not test data).
-# FUZZ_LOB excluded due to known RAC phantom transaction bugs (olr#26, olr#10).
-EXCLUDED_TABLES = {'FUZZ_STATS', 'FUZZ_LOB'}
+# Tables always excluded from comparison (stats/bookkeeping, not test data).
+# Additional tables can be excluded via --exclude-tables CLI flag.
+EXCLUDED_TABLES = {'FUZZ_STATS'}
 
 
 def is_unavailable(v):
@@ -50,8 +51,9 @@ def normalize_columns(d):
     return {k: normalize_value(v) for k, v in d.items()}
 
 
-def parse_debezium_jsonl(path):
+def parse_debezium_jsonl(path, excluded_tables=None):
     """Parse a Debezium JSONL file into normalized records."""
+    skip_tables = EXCLUDED_TABLES | (excluded_tables or set())
     records = []
     with open(path) as f:
         for line in f:
@@ -71,7 +73,7 @@ def parse_debezium_jsonl(path):
                 continue
 
             # Skip excluded tables
-            if table in EXCLUDED_TABLES:
+            if table in skip_tables:
                 continue
 
             records.append({
@@ -333,12 +335,18 @@ def compare(lm_records, olr_records):
 
 
 def main():
-    if len(sys.argv) != 3:
-        print(f"Usage: {sys.argv[0]} <logminer.jsonl> <olr.jsonl>", file=sys.stderr)
-        sys.exit(2)
-
-    lm_records = parse_debezium_jsonl(sys.argv[1])
-    olr_records = parse_debezium_jsonl(sys.argv[2])
+    parser = argparse.ArgumentParser(
+        description='Compare Debezium LogMiner vs OLR adapter outputs.')
+    parser.add_argument('logminer_jsonl', help='LogMiner JSONL file')
+    parser.add_argument('olr_jsonl', help='OLR JSONL file')
+    parser.add_argument('--exclude-tables', default='',
+                        help='Comma-separated list of additional tables to exclude')
+    args = parser.parse_args()
+
+    extra_excluded = set(t.strip() for t in args.exclude_tables.split(',') if t.strip())
+
+    lm_records = parse_debezium_jsonl(args.logminer_jsonl, extra_excluded)
+    olr_records = parse_debezium_jsonl(args.olr_jsonl, extra_excluded)
 
     # Merge LogMiner's split LOB events (OLR already emits merged events)
     lm_merged = merge_lob_events(lm_records)

tests/dbz-twin/rac/Dockerfile.consumer

@@ -0,0 +1,2 @@
+FROM python:3.12-slim
+RUN pip install --no-cache-dir kafka-python-ng==2.2.3

tests/dbz-twin/rac/FUZZ-TEST.md

@@ -99,7 +99,7 @@ Exit 0 = PASS (no non-LOB mismatches), exit 1 = FAIL.
 | `./fuzz-test.sh status` | Show container status, consumer counts, OLR memory |
 | `./fuzz-test.sh validate` | Wait for consumer drain, run validator, report PASS/FAIL |
 | `./fuzz-test.sh logs <c>` | Show logs: kafka, logminer, olr, consumer, validator, olr-vm |
-| `./fuzz-test.sh down` | Stop all containers and remove volumes |
+| `./fuzz-test.sh down` | Stop all containers and remove volumes (including fuzz-data) |
 
 ## Prerequisites
 
@@ -135,8 +135,9 @@ CREATE TABLE lm_events (
 -- olr_events: identical schema
 ```
 
-The database persists after `down` is called. Query it directly for
-investigation:
+The `fuzz-data` volume is deleted by `./fuzz-test.sh down` (which runs
+`docker compose down -v`). To inspect the database **before** tearing down,
+query it while containers are still running:
 
 ```bash
 docker run --rm -v rac_fuzz-data:/data python:3.12-slim python3 -c "

tests/dbz-twin/rac/docker-compose-fuzz.yaml

@@ -49,7 +49,10 @@ services:
       - dbz-olr-data:/debezium/data
 
   consumer:
-    image: python:3.12-slim
+    build:
+      context: .
+      dockerfile: Dockerfile.consumer
+    image: fuzz-consumer:latest
     container_name: fuzz-consumer
     network_mode: host
     depends_on:

tests/dbz-twin/rac/fuzz-test.sh

@@ -237,12 +237,19 @@ SQL
     done
     echo ""
 
-    wait $pid1 || true
-    wait $pid2 || true
+    local rc1=0 rc2=0
+    wait $pid1 || rc1=$?
+    wait $pid2 || rc2=$?
 
     echo "  Node 1: $(grep 'FUZZ_DONE:' "$work_dir/fuzz_out1.log" || echo 'no output')"
     echo "  Node 2: $(grep 'FUZZ_DONE:' "$work_dir/fuzz_out2.log" || echo 'no output')"
 
+    if [[ $rc1 -ne 0 || $rc2 -ne 0 ]]; then
+        echo "ERROR: fuzz workload failed on one or more RAC nodes (rc1=$rc1, rc2=$rc2)" >&2
+        echo "  Check logs: $work_dir/fuzz_out1.log, $work_dir/fuzz_out2.log" >&2
+        exit 1
+    fi
+
     # Flush redo
     _exec_sysdba "$work_dir/log_switch.sql" > /dev/null
     sleep 3
@@ -306,8 +313,8 @@ action_validate() {
     echo "  Consumer drained (LM events: $cur_count)"
 
     # Start validator (uses 'validate' profile)
-    docker compose -f "$COMPOSE_FILE" run --rm validator
-    local exit_code=$?
+    local exit_code=0
+    docker compose -f "$COMPOSE_FILE" run --rm validator || exit_code=$?
     echo ""
     echo "  OLR memory: $(_olr_memory_mb) MB"
 

tests/dbz-twin/rac/kafka-consumer.py

@@ -16,13 +16,7 @@
 import sys
 import time
 
-# kafka-python-ng imported after install check
-try:
-    from kafka import KafkaConsumer
-except ImportError:
-    print("Installing kafka-python-ng...", flush=True)
-    os.system("pip install -q kafka-python-ng")
-    from kafka import KafkaConsumer
+from kafka import KafkaConsumer
 
 KAFKA_BOOTSTRAP = os.environ.get('KAFKA_BOOTSTRAP', 'localhost:9092')
 SQLITE_DB = os.environ.get('SQLITE_DB', '/app/data/fuzz.db')
@@ -158,9 +152,11 @@ def main():
     consumer.poll(timeout_ms=1000)
     print(f"Subscribed to {LM_TOPIC} and {OLR_TOPIC}", flush=True)
 
-    # Track per-event_id sequence numbers for LOB split handling
+    # Track per-event_id sequence numbers for LOB split handling.
+    # Periodically trimmed to avoid unbounded growth during long runs.
     lm_seq = {}   # event_id -> next seq
     olr_seq = {}  # event_id -> next seq
+    SEQ_TRIM_THRESHOLD = 10000  # Trim when maps exceed this size
 
     lm_count = 0
     olr_count = 0
@@ -222,6 +218,16 @@ def main():
                 batch = []
                 batch_start = time.time()
 
+            # Trim seq maps to bound memory during long runs.
+            # Only seq > 0 matters (LOB splits); most events have seq=0 and
+            # can be safely evicted since they won't be seen again.
+            for seq_map in (lm_seq, olr_seq):
+                if len(seq_map) > SEQ_TRIM_THRESHOLD:
+                    # Keep only entries with seq > 0 (active LOB splits)
+                    to_delete = [k for k, v in seq_map.items() if v <= 1]
+                    for k in to_delete:
+                        del seq_map[k]
+
             # Report progress every 30 seconds
             now = time.time()
             if now - last_report >= 30: