feat: hybrid 3-connector fuzz test with patched Debezium image

Switch to rophy/debezium-server:3.5.0-2a7978c0af which includes:
- debezium-config JAR fix (missing in stock 3.5.0.Final)
- mergeLogsByPrecedence RAC fix (thread-aware dedup)

Implement hybrid architecture: OLR for non-LOB tables +
LogMiner for LOB tables on the "actual" side.

- Add dbz-lob-logminer service (LOB-only LogMiner connector)
- OLR config: skip-lob-tables=1
- Consumer: subscribe to 3 topics, route olr-lob-events to OLR side
- Validator: remove KNOWN_LOB_TABLES exemption, LOB mismatches
  are real failures with hybrid setup
- fuzz-test.sh: 3-connector offset seeding, wait, logs
- Remove restart: unless-stopped, remove Beta1 JAR mounts

Author: rophy

tests/dbz-twin/rac/config/application-lob-logminer-kafka.properties

@@ -0,0 +1,43 @@
+# Debezium Server — LogMiner adapter → Kafka sink (LOB tables only)
+# Complements OLR by handling LOB tables that OLR skips (skip-lob-tables=1).
+# Events route to olr-lob-events topic and merge into OLR's "actual" stream.
+quarkus.http.port=8083
+debezium.sink.type=kafka
+debezium.sink.kafka.producer.bootstrap.servers=localhost:9092
+debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer
+
+debezium.format.value=json
+debezium.format.value.schemas.enable=false
+debezium.format.key=json
+debezium.format.key.schemas.enable=false
+
+debezium.source.connector.class=io.debezium.connector.oracle.OracleConnector
+debezium.source.database.connection.adapter=logminer
+debezium.source.database.hostname=${VM_HOST}
+debezium.source.database.port=1521
+debezium.source.database.user=c##dbzuser
+debezium.source.database.password=dbz
+debezium.source.database.dbname=ORCLCDB
+debezium.source.database.pdb.name=ORCLPDB
+debezium.source.topic.prefix=olr-lob
+debezium.source.schema.include.list=OLR_TEST
+debezium.source.table.include.list=OLR_TEST.FUZZ_LOB
+debezium.source.snapshot.mode=recovery
+debezium.source.log.mining.strategy=online_catalog
+debezium.source.lob.enabled=true
+
+# Route all tables to a single topic for ordered delivery
+debezium.transforms=route
+debezium.transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
+debezium.transforms.route.regex=.*
+debezium.transforms.route.replacement=olr-lob-events
+
+debezium.source.offset.storage=org.apache.kafka.connect.storage.KafkaOffsetBackingStore
+debezium.source.offset.storage.topic=dbz-olr-lob-offsets
+debezium.source.offset.storage.partitions=1
+debezium.source.offset.storage.replication.factor=1
+debezium.source.bootstrap.servers=localhost:9092
+debezium.source.offset.flush.interval.ms=0
+debezium.source.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
+debezium.source.schema.history.internal.file.filename=/debezium/data/schema-history.dat

tests/dbz-twin/rac/config/olr-config.json

@@ -27,7 +27,8 @@
         "scn-type": 1,
         "timestamp-type": 1,
         "user-type": 0,
-        "redo-thread": 0
+        "redo-thread": 0,
+        "skip-lob-tables": 1
       },
       "filter": {
         "table": [

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

@@ -22,9 +22,11 @@ services:
       start_period: 15s
 
   dbz-logminer:
-    image: quay.io/debezium/server:3.5.0.Beta1
+    image: rophy/debezium-server:3.5.0-2a7978c0af
     container_name: fuzz-dbz-logminer
     network_mode: host
+    environment:
+      VM_HOST: ${VM_HOST:?VM_HOST is required}
     depends_on:
       kafka:
         condition: service_healthy
@@ -34,19 +36,33 @@ services:
       - dbz-logminer-data:/debezium/data
 
   dbz-olr:
-    image: quay.io/debezium/server:3.5.0.Beta1
+    image: rophy/debezium-server:3.5.0-2a7978c0af
     container_name: fuzz-dbz-olr
     network_mode: host
-    restart: unless-stopped
+    environment:
+      VM_HOST: ${VM_HOST:?VM_HOST is required}
     depends_on:
       kafka:
         condition: service_healthy
     volumes:
       - ./config/application-olr-kafka.properties:/debezium/config/application.properties:ro
       - ../lib/ojdbc8.jar:/debezium/lib/ojdbc8.jar:ro
-      - ../lib/debezium-connector-oracle-3.5.0.Beta1.jar:/debezium/lib/debezium-connector-oracle-3.5.0.Beta1.jar:ro
       - dbz-olr-data:/debezium/data
 
+  dbz-lob-logminer:
+    image: rophy/debezium-server:3.5.0-2a7978c0af
+    container_name: fuzz-dbz-lob-logminer
+    network_mode: host
+    environment:
+      VM_HOST: ${VM_HOST:?VM_HOST is required}
+    depends_on:
+      kafka:
+        condition: service_healthy
+    volumes:
+      - ./config/application-lob-logminer-kafka.properties:/debezium/config/application.properties:ro
+      - ../lib/ojdbc8.jar:/debezium/lib/ojdbc8.jar:ro
+      - dbz-lob-logminer-data:/debezium/data
+
   consumer:
     build:
       context: .
@@ -63,6 +79,7 @@ services:
       SQLITE_DB: /app/data/fuzz.db
       LM_TOPIC: lm-events
       OLR_TOPIC: olr-events
+      OLR_LOB_TOPIC: olr-lob-events
     volumes:
       - ./kafka-consumer.py:/app/kafka-consumer.py:ro
       - fuzz-data:/app/data
@@ -90,4 +107,5 @@ services:
 volumes:
   dbz-logminer-data:
   dbz-olr-data:
+  dbz-lob-logminer-data:
   fuzz-data:

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

@@ -84,7 +84,7 @@ _seed_debezium_offsets() {
 
     # topic_prefix matches debezium.source.topic.prefix in each connector config
     # offset_topic matches debezium.source.offset.storage.topic
-    local -A topics=( [logminer]=dbz-lm-offsets [olr]=dbz-olr-offsets )
+    local -A topics=( [logminer]=dbz-lm-offsets [olr]=dbz-olr-offsets [olr-lob]=dbz-olr-lob-offsets )
     for topic_prefix in "${!topics[@]}"; do
         local offset_topic="${topics[$topic_prefix]}"
         local offset_key="[\"kafka\",{\"server\":\"${topic_prefix}\"}]"
@@ -212,15 +212,16 @@ action_up() {
 
     # Wait for Debezium connectors
     echo "  Waiting for Debezium connectors..."
-    for i in $(seq 1 60); do
-        LM_OK=false; OLR_OK=false
-        docker logs fuzz-dbz-logminer 2>&1 | tail -10 | grep -q "Starting streaming" && LM_OK=true
-        docker logs fuzz-dbz-olr 2>&1 | tail -10 | grep -q "streaming client started\|Starting streaming" && OLR_OK=true
-        if $LM_OK && $OLR_OK; then
-            echo "  Debezium: ready"
+    for i in $(seq 1 90); do
+        LM_OK=false; OLR_OK=false; LOB_LM_OK=false
+        docker logs fuzz-dbz-logminer 2>&1 | grep -q "Starting streaming" && LM_OK=true
+        docker logs fuzz-dbz-olr 2>&1 | grep -q "streaming client started\|Starting streaming" && OLR_OK=true
+        docker logs fuzz-dbz-lob-logminer 2>&1 | grep -q "Starting streaming" && LOB_LM_OK=true
+        if $LM_OK && $OLR_OK && $LOB_LM_OK; then
+            echo "  Debezium: ready (3 connectors)"
             break
         fi
-        [[ $i -eq 60 ]] && { echo "ERROR: Debezium connectors did not start" >&2; exit 1; }
+        [[ $i -eq 90 ]] && { echo "ERROR: Debezium connectors did not start" >&2; exit 1; }
         sleep 2
     done
 
@@ -427,19 +428,20 @@ action_validate() {
 action_logs() {
     local component="${1:-}"
     case "$component" in
-        kafka)      docker logs fuzz-kafka 2>&1 ;;
-        logminer)   docker logs fuzz-dbz-logminer 2>&1 ;;
-        olr)        docker logs fuzz-dbz-olr 2>&1 ;;
-        consumer)   docker logs fuzz-consumer 2>&1 ;;
-        validator)  docker logs fuzz-validator 2>&1 ;;
-        olr-vm)     ssh $_SSH_OPTS "${VM_USER}@${VM_HOST}" "podman logs $OLR_CONTAINER" 2>/dev/null ;;
+        kafka)          docker logs fuzz-kafka 2>&1 ;;
+        logminer)       docker logs fuzz-dbz-logminer 2>&1 ;;
+        olr)            docker logs fuzz-dbz-olr 2>&1 ;;
+        lob-logminer)   docker logs fuzz-dbz-lob-logminer 2>&1 ;;
+        consumer)       docker logs fuzz-consumer 2>&1 ;;
+        validator)      docker logs fuzz-validator 2>&1 ;;
+        olr-vm)         ssh $_SSH_OPTS "${VM_USER}@${VM_HOST}" "podman logs $OLR_CONTAINER" 2>/dev/null ;;
         "")
             echo "Usage: $0 logs <component>"
-            echo "Components: kafka, logminer, olr, consumer, validator, olr-vm"
+            echo "Components: kafka, logminer, olr, lob-logminer, consumer, validator, olr-vm"
             ;;
         *)
             echo "Unknown component: $component" >&2
-            echo "Components: kafka, logminer, olr, consumer, validator, olr-vm"
+            echo "Components: kafka, logminer, olr, lob-logminer, consumer, validator, olr-vm"
             exit 1
             ;;
     esac

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

@@ -94,13 +94,18 @@ def extract_event_info(event):
 
 LM_TOPIC = os.environ.get('LM_TOPIC', 'lm-events')
 OLR_TOPIC = os.environ.get('OLR_TOPIC', 'olr-events')
+OLR_LOB_TOPIC = os.environ.get('OLR_LOB_TOPIC', 'olr-lob-events')
 
 
 def determine_adapter(topic):
-    """Determine adapter (logminer or olr) from Kafka topic name."""
+    """Determine adapter (logminer or olr) from Kafka topic name.
+
+    The OLR LOB topic (LogMiner for LOB tables) is treated as 'olr' because
+    it complements OLR on the "actual" side of the comparison.
+    """
     if topic == LM_TOPIC:
         return 'logminer'
-    elif topic == OLR_TOPIC:
+    elif topic in (OLR_TOPIC, OLR_LOB_TOPIC):
         return 'olr'
     # Fallback for per-table topics
     if topic.startswith('logminer'):
@@ -139,18 +144,19 @@ def main():
                 sys.exit(1)
 
     # Wait for topics to appear, then subscribe
-    print(f"Waiting for topics: {LM_TOPIC}, {OLR_TOPIC}...", flush=True)
+    all_topics = [LM_TOPIC, OLR_TOPIC, OLR_LOB_TOPIC]
+    print(f"Waiting for topics: {', '.join(all_topics)}...", flush=True)
     for attempt in range(60):
         topics = consumer.topics()
         if LM_TOPIC in topics or OLR_TOPIC in topics:
-            print(f"  Found topics: {[t for t in (LM_TOPIC, OLR_TOPIC) if t in topics]}", flush=True)
+            print(f"  Found topics: {[t for t in all_topics if t in topics]}", flush=True)
             break
         time.sleep(5)
 
-    consumer.subscribe([LM_TOPIC, OLR_TOPIC])
+    consumer.subscribe(all_topics)
     # Force metadata refresh
     consumer.poll(timeout_ms=1000)
-    print(f"Subscribed to {LM_TOPIC} and {OLR_TOPIC}", flush=True)
+    print(f"Subscribed to {', '.join(all_topics)}", flush=True)
 
     # Track per-event_id sequence numbers for LOB split handling.
     lm_seq = {}   # event_id -> next seq

tests/dbz-twin/rac/validator.py

@@ -24,9 +24,10 @@
 POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '10'))
 IDLE_TIMEOUT = int(os.environ.get('IDLE_TIMEOUT', '120'))
 
-# Known LOB phantom transaction issues (olr#26, olr#10)
-# These produce expected mismatches — report but don't fail
-KNOWN_LOB_TABLES = {'FUZZ_LOB'}
+# LOB tables that use final-state replay for comparison.
+# With the hybrid setup (OLR for non-LOB + LogMiner for LOB), these tables
+# should match exactly. Mismatches are treated as real failures.
+LOB_TABLES = {'FUZZ_LOB'}
 
 
 def normalize_value(v):
@@ -132,7 +133,6 @@ def main():
     total_validated = 0
     total_matched = 0
     total_mismatches = 0
-    total_lob_known = 0  # Known LOB issues (expected)
     total_missing_lm = 0
     total_missing_olr = 0
     total_tail_olr = 0  # OLR ahead of LM at drain time (not a bug)
@@ -245,14 +245,12 @@ def main():
                         "SELECT table_name FROM olr_events WHERE event_id = ? LIMIT 1",
                         (eid,)).fetchone()
                 event_table = tbl_row['table_name'] if tbl_row else '?'
-                is_lob = event_table in KNOWN_LOB_TABLES
+                is_lob = event_table in LOB_TABLES
 
                 if in_lm and not in_olr:
                     total_missing_olr += 1
                     if is_tail:
                         total_tail_lm += 1
-                    elif is_lob:
-                        total_lob_known += 1
                     else:
                         total_mismatches += 1
                         print(f"[MISSING_OLR] {eid} ({event_table})", flush=True)
@@ -263,8 +261,6 @@ def main():
                     total_missing_lm += 1
                     if is_tail:
                         total_tail_olr += 1
-                    elif is_lob:
-                        total_lob_known += 1
                     else:
                         total_mismatches += 1
                         print(f"[EXTRA_OLR] {eid} ({event_table})", flush=True)
@@ -283,20 +279,26 @@ def main():
 
                 if is_lob:
                     # LOB tables: replay ops into final state, compare end result.
-                    # LogMiner merges INSERT + LOB_WRITE into a single record (L2),
-                    # and OLR may have extra/fewer intermediate events due to
-                    # phantom undo (#15). Comparing final state avoids both issues.
+                    # Both sides use LogMiner (expected=full LM, actual=LOB-only LM),
+                    # so they should produce identical final state.
                     lm_state, lm_exists = replay_final_state(lm_recs)
                     olr_state, olr_exists = replay_final_state(olr_recs)
 
                     if lm_exists != olr_exists:
-                        total_lob_known += 1
+                        total_mismatches += 1
+                        print(f"[LOB_EXISTENCE] {eid} ({event_table}): "
+                              f"LM exists={lm_exists} OLR exists={olr_exists}",
+                              flush=True)
                         total_validated += 1
                     else:
                         diffs = compare_values(lm_state, olr_state,
                                                event_table, 'after')
                         if diffs:
-                            total_lob_known += 1
+                            total_mismatches += 1
+                            print(f"[LOB_VALUE_DIFF] {eid} ({event_table}):",
+                                  flush=True)
+                            for d in diffs[:5]:
+                                print(d, flush=True)
                         else:
                             total_matched += 1
                         total_validated += 1
@@ -373,7 +375,7 @@ def main():
             tail_str = (f" tail_olr={total_tail_olr} tail_lm={total_tail_lm}"
                         if total_tail_olr or total_tail_lm else "")
             print(f"[validator] validated={total_validated} matched={total_matched} "
-                  f"mismatches={total_mismatches} lob_known={total_lob_known} "
+                  f"mismatches={total_mismatches} "
                   f"missing_olr={total_missing_olr} extra_olr={total_missing_lm}"
                   f"{tail_str} "
                   f"lm_total={lm_count} olr_total={olr_count} "
@@ -391,7 +393,6 @@ def main():
     print(f"  Total validated:    {total_validated}", flush=True)
     print(f"  Matched:            {total_matched}", flush=True)
     print(f"  Mismatches:         {total_mismatches}", flush=True)
-    print(f"  LOB known issues:   {total_lob_known}", flush=True)
     print(f"  Missing from OLR:   {total_missing_olr}", flush=True)
     print(f"  Extra in OLR:       {total_missing_lm}", flush=True)
     if total_tail_olr or total_tail_lm:
@@ -404,13 +405,8 @@ def main():
         sys.exit(1)
     else:
         print("\n  RESULT: PASS", flush=True)
-        qualifiers = []
-        if total_lob_known > 0:
-            qualifiers.append(f"{total_lob_known} known LOB issues")
         if total_tail_olr + total_tail_lm > 0:
-            qualifiers.append(f"{total_tail_olr + total_tail_lm} tail events")
-        if qualifiers:
-            print(f"  ({', '.join(qualifiers)})", flush=True)
+            print(f"  ({total_tail_olr + total_tail_lm} tail events)", flush=True)
         sys.exit(0)