Add integration tests for Apache Pulsar connection validation using Testcontainers

Signed-off-by: Philippe Camus <pxcamus@pm.me>

Author: pxcamus

debezium-platform-conductor/pom.xml

@@ -447,6 +447,12 @@
             <artifactId>testcontainers-mssqlserver</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers-pulsar</artifactId>
+            <scope>test</scope>
+        </dependency>
+
 
         <dependency>
             <groupId>ch.qos.logback</groupId>

debezium-platform-conductor/src/test/java/io/debezium/platform/environment/connection/PulsarConnectionValidatorIT.java

@@ -0,0 +1,52 @@
+package io.debezium.platform.environment.connection;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.pulsar.PulsarContainer;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import io.debezium.platform.data.dto.ConnectionValidationResult;
+import io.debezium.platform.data.model.ConnectionEntity;
+import io.debezium.platform.domain.views.Connection;
+import io.debezium.platform.environment.connection.destination.PulsarConnectionValidator;
+import io.debezium.platform.environment.destination.ApachePulsarTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+
+@QuarkusTest
+@QuarkusTestResource(value = ApachePulsarTestResource.class, restrictToAnnotatedClass = true)
+class PulsarConnectionValidatorIT {
+    public static final int DEFAULT_30_SECONDS_TIMEOUT = 30;
+
+    private PulsarConnectionValidator validator;
+
+    @BeforeEach
+    void setUp() {
+        validator = new PulsarConnectionValidator(DEFAULT_30_SECONDS_TIMEOUT);
+    }
+
+    @Test
+    @DisplayName("Should successfully validate connection with valid Pulsar configuration")
+    void shouldValidateSuccessfulConnection() {
+        PulsarContainer container = ApachePulsarTestResource.getContainer();
+
+        Awaitility.await()
+                .atMost(300, TimeUnit.SECONDS)
+                .until(container::isRunning);
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("serviceHttpUrl", container.getHttpServiceUrl());
+        Connection connection = new TestConnectionView(ConnectionEntity.Type.APACHE_PULSAR, config);
+
+        ConnectionValidationResult result = validator.validate(connection);
+
+        assertTrue(result.valid(), "Connection validation should succeed");
+    }
+}

debezium-platform-conductor/src/test/java/io/debezium/platform/environment/destination/ApachePulsarTestResource.java

@@ -0,0 +1,31 @@
+package io.debezium.platform.environment.destination;
+
+import java.util.Map;
+
+import org.testcontainers.pulsar.PulsarContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+
+public class ApachePulsarTestResource implements QuarkusTestResourceLifecycleManager {
+    private static final PulsarContainer PULSAR = new PulsarContainer(DockerImageName.parse("apachepulsar/pulsar:4.1.3"));
+
+    static {
+        PULSAR.withExposedPorts(8080);
+    }
+
+    public static PulsarContainer getContainer() {
+        return PULSAR;
+    }
+
+    @Override
+    public Map<String, String> start() {
+        PULSAR.start();
+        return Map.of();
+    }
+
+    @Override
+    public void stop() {
+        PULSAR.stop();
+    }
+}