From 769060f29258310f8e2875a5f2dcc6dbefee8e55 Mon Sep 17 00:00:00 2001 From: Fiore Mario Vitale Date: Tue, 29 Sep 2026 14:57:41 +0200 Subject: [PATCH] debezium/dbz#2732 Correctly discover running engine instance Signed-off-by: Fiore Mario Vitale (cherry picked from commit 5a3fac00dea892bc7141c2fe4821b5d0abe69281) --- .../server/rest/signal/SignalResource.java | 41 ++++-- .../io/debezium/server/DebeziumServerIT.java | 15 +- .../rest/signal/SignalResourceTest.java | 130 ++++++++++++++++++ 3 files changed, 169 insertions(+), 17 deletions(-) create mode 100644 debezium-server-core/src/test/java/io/debezium/server/rest/signal/SignalResourceTest.java diff --git a/debezium-server-core/src/main/java/io/debezium/server/rest/signal/SignalResource.java b/debezium-server-core/src/main/java/io/debezium/server/rest/signal/SignalResource.java index e649342c8..e9a1afb5e 100644 --- a/debezium-server-core/src/main/java/io/debezium/server/rest/signal/SignalResource.java +++ b/debezium-server-core/src/main/java/io/debezium/server/rest/signal/SignalResource.java @@ -5,15 +5,16 @@ */ 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; @@ -21,27 +22,47 @@ public class SignalResource { private final DebeziumServerConfig config; - private final DebeziumConnectorRegistry registry; + private final DebeziumConnectorsRegistry registry; @Inject - public SignalResource(DebeziumServerConfig config, Instance 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 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()); } diff --git a/debezium-server-core/src/test/java/io/debezium/server/DebeziumServerIT.java b/debezium-server-core/src/test/java/io/debezium/server/DebeziumServerIT.java index bc31f8e45..ce53b5e57 100644 --- a/debezium-server-core/src/test/java/io/debezium/server/DebeziumServerIT.java +++ b/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 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( diff --git a/debezium-server-core/src/test/java/io/debezium/server/rest/signal/SignalResourceTest.java b/debezium-server-core/src/test/java/io/debezium/server/rest/signal/SignalResourceTest.java new file mode 100644 index 000000000..80876ef2d --- /dev/null +++ b/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}. + * + *

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; + } +}