debezium/dbz#2732 Correctly discover running engine instance

debezium-server-core/src/main/java/io/debezium/server/rest/signal/SignalResource.java

@@ -5,43 +5,64 @@
  */
 package io.debezium.server.rest.signal;
 
-import jakarta.enterprise.inject.Instance;
+import java.util.Optional;
+
 import jakarta.inject.Inject;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.core.Response;
 
-import io.debezium.DebeziumException;
 import io.debezium.engine.DebeziumEngine;
-import io.debezium.runtime.DebeziumConnectorRegistry;
+import io.debezium.runtime.Debezium;
+import io.debezium.runtime.DebeziumConnectorsRegistry;
 import io.debezium.runtime.EngineManifest;
 import io.debezium.server.configuration.DebeziumServerConfig;
 
 @Path("/signals")
 public class SignalResource {
 
     private final DebeziumServerConfig config;
-    private final DebeziumConnectorRegistry registry;
+    private final DebeziumConnectorsRegistry registry;
 
     @Inject
-    public SignalResource(DebeziumServerConfig config, Instance<DebeziumConnectorRegistry> instance) {
+    public SignalResource(DebeziumServerConfig config, DebeziumConnectorsRegistry registry) {
         this.config = config;
-        this.registry = instance.stream().findFirst().orElseThrow(() -> new DebeziumException("No DebeziumConnectorRegistry available"));
+        this.registry = registry;
     }
 
     @POST
     public Response post(DSSignal dsSignal) {
-        var signaler = registry.get(new EngineManifest("default")).signaler();
+        if (!config.api().enabled()) {
+            return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
+        }
 
-        if (signaler == null || !config.api().enabled()) {
+        var signaler = runningEngine()
+                .map(Debezium::signaler)
+                .orElse(null);
+
+        if (signaler == null) {
             return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
         }
 
-        var signal = toSignal(dsSignal);
-        signaler.signal(signal);
+        signaler.signal(toSignal(dsSignal));
         return Response.accepted().build();
     }
 
+    /**
+     * Every Debezium Quarkus connector extension on the classpath declares an engine for the default
+     * manifest, but Debezium Server only runs the one matching the configured {@code connector.class}.
+     * Looking the engine up by manifest would therefore resolve an engine that never started, so the
+     * running engines are the ones to search.
+     *
+     * @return the running engine, or empty when no engine is running yet
+     */
+    private Optional<Debezium> runningEngine() {
+        return registry.runningEngines()
+                .stream()
+                .filter(engine -> EngineManifest.DEFAULT.equals(engine.manifest()))
+                .findFirst();
+    }
+
     private DebeziumEngine.Signal toSignal(DSSignal dsSignal) {
         return new DebeziumEngine.Signal(dsSignal.id(), dsSignal.type(), dsSignal.data(), dsSignal.additionalData());
     }

debezium-server-core/src/test/java/io/debezium/server/DebeziumServerIT.java

@@ -11,7 +11,6 @@
 import java.time.Duration;
 
 import jakarta.enterprise.event.Observes;
-import jakarta.enterprise.inject.Instance;
 import jakarta.inject.Inject;
 import jakarta.ws.rs.core.Response;
 
@@ -23,9 +22,9 @@
 
 import io.debezium.config.CommonConnectorConfig;
 import io.debezium.engine.DebeziumEngine;
-import io.debezium.runtime.DebeziumConnectorRegistry;
+import io.debezium.runtime.Debezium;
+import io.debezium.runtime.DebeziumConnectorsRegistry;
 import io.debezium.runtime.DebeziumStatus;
-import io.debezium.runtime.EngineManifest;
 import io.debezium.runtime.events.ConnectorStartedEvent;
 import io.debezium.runtime.events.DebeziumCompletionEvent;
 import io.debezium.testing.testcontainers.PostgresTestResourceLifecycleManager;
@@ -55,7 +54,7 @@ public class DebeziumServerIT {
     TestConsumer testConsumer;
 
     @Inject
-    Instance<DebeziumConnectorRegistry> connectorRegistries;
+    DebeziumConnectorsRegistry connectorsRegistry;
 
     @Inject
     DebeziumMetrics metrics;
@@ -110,10 +109,12 @@ public void testDebeziumMetricsWithPostgres() {
     public void testDebeziumServerSignals() {
         Testing.Print.enable();
 
-        // wait for the connector to start
+        // wait for the connector to start; each connector extension on the classpath contributes a
+        // registry, but only the extension matching the configured connector.class runs an engine
         Awaitility.await().atMost(Duration.ofSeconds(TestConfigSource.waitForSeconds()))
-                .untilAsserted(() -> Assertions.assertThat(connectorRegistries.stream().findFirst().get().get(new EngineManifest("default")).status())
-                        .isEqualTo(new DebeziumStatus(DebeziumStatus.State.POLLING)));
+                .untilAsserted(() -> Assertions.assertThat(connectorsRegistry.runningEngines())
+                        .extracting(Debezium::status)
+                        .containsExactly(new DebeziumStatus(DebeziumStatus.State.POLLING)));
 
         // prepare signal
         var signal = new DebeziumEngine.Signal(

debezium-server-core/src/test/java/io/debezium/server/rest/signal/SignalResourceTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.server.rest.signal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Optional;
+
+import jakarta.ws.rs.core.Response;
+
+import org.junit.jupiter.api.Test;
+
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.runtime.Debezium;
+import io.debezium.runtime.DebeziumConnectorsRegistry;
+import io.debezium.runtime.EngineManifest;
+import io.debezium.server.configuration.DebeziumServerConfig;
+
+/**
+ * Unit tests for {@link SignalResource}.
+ *
+ * <p>Debezium Server bundles one Debezium Quarkus connector extension per supported connector and each
+ * of them declares an engine for {@link EngineManifest#DEFAULT}. Only the engine whose connector
+ * matches the configured {@code connector.class} is started, so an engine found by manifest is not
+ * necessarily an engine that can be signaled.
+ */
+public class SignalResourceTest {
+
+    private static final DSSignal DS_SIGNAL = new DSSignal("1", "log", "{\"message\": \"Signal message\"}", null);
+    private static final DebeziumEngine.Signal SIGNAL = new DebeziumEngine.Signal(DS_SIGNAL.id(), DS_SIGNAL.type(), DS_SIGNAL.data(),
+            DS_SIGNAL.additionalData());
+
+    @Test
+    public void shouldSignalTheRunningEngine() {
+        DebeziumEngine.Signaler signaler = mock(DebeziumEngine.Signaler.class);
+        DebeziumConnectorsRegistry registry = registryWithRunningEngine(engine(signaler));
+
+        SignalResource resource = new SignalResource(apiEnabled(true), registry);
+
+        Response response = resource.post(DS_SIGNAL);
+
+        assertThat(response.getStatus()).isEqualTo(Response.Status.ACCEPTED.getStatusCode());
+        verify(signaler).signal(SIGNAL);
+    }
+
+    @Test
+    public void shouldNotSignalAnEngineThatIsNotRunning() {
+        DebeziumEngine.Signaler signaler = mock(DebeziumEngine.Signaler.class);
+        DebeziumConnectorsRegistry registry = registryWithStoppedEngine(engine(signaler));
+
+        SignalResource resource = new SignalResource(apiEnabled(true), registry);
+
+        Response response = resource.post(DS_SIGNAL);
+
+        assertThat(response.getStatus()).isEqualTo(Response.Status.SERVICE_UNAVAILABLE.getStatusCode());
+        verifyNoInteractions(signaler);
+    }
+
+    @Test
+    public void shouldReturnServiceUnavailableWhenNoEngineIsRunning() {
+        DebeziumConnectorsRegistry registry = mock(DebeziumConnectorsRegistry.class);
+        when(registry.runningEngines()).thenReturn(List.of());
+        when(registry.get(EngineManifest.DEFAULT)).thenReturn(Optional.empty());
+
+        SignalResource resource = new SignalResource(apiEnabled(true), registry);
+
+        Response response = resource.post(DS_SIGNAL);
+
+        assertThat(response.getStatus()).isEqualTo(Response.Status.SERVICE_UNAVAILABLE.getStatusCode());
+    }
+
+    @Test
+    public void shouldReturnServiceUnavailableWhenApiIsDisabled() {
+        DebeziumEngine.Signaler signaler = mock(DebeziumEngine.Signaler.class);
+        DebeziumConnectorsRegistry registry = registryWithRunningEngine(engine(signaler));
+
+        SignalResource resource = new SignalResource(apiEnabled(false), registry);
+
+        Response response = resource.post(DS_SIGNAL);
+
+        assertThat(response.getStatus()).isEqualTo(Response.Status.SERVICE_UNAVAILABLE.getStatusCode());
+        verifyNoInteractions(signaler);
+    }
+
+    private DebeziumServerConfig apiEnabled(boolean enabled) {
+        DebeziumServerConfig.Api api = mock(DebeziumServerConfig.Api.class);
+        when(api.enabled()).thenReturn(enabled);
+
+        DebeziumServerConfig config = mock(DebeziumServerConfig.class);
+        when(config.api()).thenReturn(api);
+
+        return config;
+    }
+
+    private DebeziumConnectorsRegistry registryWithRunningEngine(Debezium engine) {
+        DebeziumConnectorsRegistry registry = mock(DebeziumConnectorsRegistry.class);
+        when(registry.runningEngines()).thenReturn(List.of(engine));
+        when(registry.get(EngineManifest.DEFAULT)).thenReturn(Optional.of(engine));
+
+        return registry;
+    }
+
+    /**
+     * An engine declared by a connector extension that the engine filter strategy did not select: it
+     * is resolvable by manifest, but it never started and signaling it would go nowhere.
+     */
+    private DebeziumConnectorsRegistry registryWithStoppedEngine(Debezium engine) {
+        DebeziumConnectorsRegistry registry = mock(DebeziumConnectorsRegistry.class);
+        when(registry.runningEngines()).thenReturn(List.of());
+        when(registry.get(EngineManifest.DEFAULT)).thenReturn(Optional.of(engine));
+
+        return registry;
+    }
+
+    private Debezium engine(DebeziumEngine.Signaler signaler) {
+        Debezium engine = mock(Debezium.class);
+        when(engine.manifest()).thenReturn(EngineManifest.DEFAULT);
+        when(engine.signaler()).thenReturn(signaler);
+
+        return engine;
+    }
+}