debezium/dbz#2026 Support multiengine in QuarkusHeartbeatEmitter

Signed-off-by: nonononoonononon <yuangli971@gmail.com>

Author: yuanglili

debezium-quarkus-parent/debezium-quarkus-engine-parent/runtime/src/main/java/io/quarkus/debezium/heartbeat/QuarkusHeartbeatEmitter.java

@@ -49,13 +49,18 @@ public QuarkusHeartbeatEmitter(List<DebeziumConnectorRegistry> registries, Event
 
     @Override
     public void emit(Map<String, ?> partition, OffsetContext offset) {
-        DefaultEngine.Literal.selectDefault(heartbeat.select(Engine.Literal.of(context().manifest().id())),
-                context().manifest())
-                .fire(new DebeziumHeartbeat(
-                        this.registries.getFirst().engines().getFirst().connector(),
-                        this.registries.getFirst().engines().getFirst().status(),
-                        partition,
-                        offset.getOffset()));
+        this.registries
+                .stream()
+                .flatMap(registry -> registry.engines().stream())
+                .filter(engine -> engine.manifest().equals(context().manifest()))
+                .map(engine -> Map.entry(engine, DefaultEngine.Literal.selectDefault(
+                        heartbeat.select(Engine.Literal.of(engine.manifest().id())), engine.manifest())))
+                .forEach(entry -> entry.getValue().fire(
+                        new DebeziumHeartbeat(
+                                entry.getKey().connector(),
+                                entry.getKey().status(),
+                                partition,
+                                offset.getOffset())));
     }
 
     @Override

debezium-quarkus-parent/debezium-quarkus-engine-parent/runtime/src/test/java/io/quarkus/debezium/heartbeat/QuarkusHeartbeatEmitterTest.java

@@ -7,11 +7,13 @@
 package io.quarkus.debezium.heartbeat;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.time.Instant;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import jakarta.enterprise.event.Event;
@@ -60,7 +62,28 @@ void shouldFireEventWhenCalled() {
         verify(event).fire(new DebeziumHeartbeat(CONNECTOR, DEBEZIUM_STATUS, PARTITION, Map.of("offset", "value")));
     }
 
+    @Test
+    @DisplayName("should only fire event for the engine matching the current thread context")
+    void shouldOnlyFireForEngineMatchingContext() {
+        Connector otherConnector = new Connector("other.connector");
+        DebeziumStatus otherStatus = new DebeziumStatus(DebeziumStatus.State.CREATING);
+
+        Debezium matching = generate("testing", DEBEZIUM_STATUS, CONNECTOR);
+        Debezium other = generate("alternative", otherStatus, otherConnector);
+
+        when(registry.engines()).thenReturn(List.of(matching, other));
+
+        underTest.emit(PARTITION, OFFSET);
+
+        verify(event).fire(new DebeziumHeartbeat(CONNECTOR, DEBEZIUM_STATUS, PARTITION, Map.of("offset", "value")));
+        verify(event, never()).fire(new DebeziumHeartbeat(otherConnector, otherStatus, PARTITION, Map.of("offset", "value")));
+    }
+
     public Debezium generate(DebeziumStatus status) {
+        return generate("testing", status, CONNECTOR);
+    }
+
+    public Debezium generate(String manifestId, DebeziumStatus status, Connector connector) {
         return new Debezium() {
             @Override
             public DebeziumEngine.Signaler signaler() {
@@ -79,12 +102,12 @@ public DebeziumStatus status() {
 
             @Override
             public Connector connector() {
-                return CONNECTOR;
+                return connector;
             }
 
             @Override
             public EngineManifest manifest() {
-                return null;
+                return new EngineManifest(manifestId);
             }
         };
     }