This is an automated email from the ASF dual-hosted git repository. fmariani pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit 32b77226e1ee92d409095a8f7faa287e1ce43caf Author: Croway <[email protected]> AuthorDate: Tue Mar 31 15:15:35 2026 +0200 Fix CXF/CXFRS source tests: wait for endpoint readiness and remove unused dispatch-router The CXF source connector opens the HTTP port before the service path is fully registered, causing 404 responses when the test client invokes the service immediately after port becomes available. Use Awaitility to poll the WSDL URL (CXF) or the REST endpoint (CXFRS) until the service is ready before producing test data. Also increase CXFRS test timeout from 30s to 90s. Remove unused camel-test-infra-dispatch-router dependency from both modules — the retired qdrouterd image was never used in these tests. --- .../DefaultKafkaConnectPropertyFactory.java | 1 - tests/itests-cxf/pom.xml | 14 +++++------ .../cxf/source/CamelSourceCXFITCase.java | 29 +++++++++++++++++++--- tests/itests-cxfrs/pom.xml | 14 +++++------ .../cxfrs/source/CamelSourceCXFRSITCase.java | 29 +++++++++++----------- .../sjms2/sink/CamelSinkIdempotentJMSITCase.java | 2 +- .../source/CamelSourceJMSWithAggregation.java | 2 +- 7 files changed, 55 insertions(+), 36 deletions(-) diff --git a/tests/itests-common/src/test/java/org/apache/camel/kafkaconnector/common/services/kafkaconnect/DefaultKafkaConnectPropertyFactory.java b/tests/itests-common/src/test/java/org/apache/camel/kafkaconnector/common/services/kafkaconnect/DefaultKafkaConnectPropertyFactory.java index 04fff966c5..ad88e936c4 100644 --- a/tests/itests-common/src/test/java/org/apache/camel/kafkaconnector/common/services/kafkaconnect/DefaultKafkaConnectPropertyFactory.java +++ b/tests/itests-common/src/test/java/org/apache/camel/kafkaconnector/common/services/kafkaconnect/DefaultKafkaConnectPropertyFactory.java @@ -20,7 +20,6 @@ package org.apache.camel.kafkaconnector.common.services.kafkaconnect; import java.util.Properties; import org.apache.camel.kafkaconnector.common.PluginPathHelper; -import org.apache.camel.kafkaconnector.common.utils.NetworkUtils; import org.apache.kafka.connect.runtime.rest.RestServerConfig; import org.apache.kafka.connect.runtime.standalone.StandaloneConfig; import org.slf4j.Logger; diff --git a/tests/itests-cxf/pom.xml b/tests/itests-cxf/pom.xml index 623275ea8e..e768ea080a 100644 --- a/tests/itests-cxf/pom.xml +++ b/tests/itests-cxf/pom.xml @@ -43,13 +43,6 @@ <version>${camel.version}</version> <scope>test</scope> </dependency> - <dependency> - <groupId>org.apache.camel</groupId> - <artifactId>camel-test-infra-dispatch-router</artifactId> - <version>${camel.version}</version> - <scope>test</scope> - </dependency> - <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-cxf-soap</artifactId> @@ -81,6 +74,13 @@ <artifactId>commons-io</artifactId> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>${awaitility-version}</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>javax.annotation</groupId> <artifactId>javax.annotation-api</artifactId> diff --git a/tests/itests-cxf/src/test/java/org/apache/camel/kafkaconnector/cxf/source/CamelSourceCXFITCase.java b/tests/itests-cxf/src/test/java/org/apache/camel/kafkaconnector/cxf/source/CamelSourceCXFITCase.java index b8ee82ad63..add50d0ce4 100644 --- a/tests/itests-cxf/src/test/java/org/apache/camel/kafkaconnector/cxf/source/CamelSourceCXFITCase.java +++ b/tests/itests-cxf/src/test/java/org/apache/camel/kafkaconnector/cxf/source/CamelSourceCXFITCase.java @@ -17,6 +17,10 @@ package org.apache.camel.kafkaconnector.cxf.source; +import java.net.HttpURLConnection; +import java.net.URI; +import java.time.Duration; + import org.apache.camel.kafkaconnector.common.test.CamelSourceTestSupport; import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer; import org.apache.camel.kafkaconnector.common.utils.NetworkUtils; @@ -24,6 +28,7 @@ import org.apache.camel.kafkaconnector.cxf.client.CXFServiceUtil; import org.apache.camel.kafkaconnector.cxf.common.HelloService; import org.apache.camel.test.infra.common.TestUtils; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.awaitility.Awaitility; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,16 +58,32 @@ public abstract class CamelSourceCXFITCase extends CamelSourceTestSupport { protected void produceTestData() { TestUtils.waitFor(() -> NetworkUtils.portIsOpen("localhost", PORT)); + // Wait for the CXF endpoint to be fully registered (WSDL available) + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofSeconds(1)) + .pollDelay(Duration.ZERO) + .until(() -> { + try { + HttpURLConnection conn = (HttpURLConnection) URI.create(SIMPLE_ENDPOINT_ADDRESS + "?wsdl").toURL().openConnection(); + conn.setRequestMethod("GET"); + conn.setConnectTimeout(1000); + conn.setReadTimeout(1000); + int code = conn.getResponseCode(); + conn.disconnect(); + return code == 200; + } catch (Exception e) { + return false; + } + }); + try { HelloService client = CXFServiceUtil.getService(SIMPLE_ENDPOINT_ADDRESS, HelloService.class); - for (int i = 0; i < expect; i++) { client.echo("Test message " + i); } - - } catch (Exception e) { - LOG.info("Unable to invoke service: {}", e.getMessage(), e); + LOG.error("Unable to invoke service: {}", e.getMessage(), e); fail("Unable to invoke service"); } } diff --git a/tests/itests-cxfrs/pom.xml b/tests/itests-cxfrs/pom.xml index b32668d25c..6ce096ad91 100644 --- a/tests/itests-cxfrs/pom.xml +++ b/tests/itests-cxfrs/pom.xml @@ -43,13 +43,6 @@ <version>${camel.version}</version> <scope>test</scope> </dependency> - <dependency> - <groupId>org.apache.camel</groupId> - <artifactId>camel-test-infra-dispatch-router</artifactId> - <version>${camel.version}</version> - <scope>test</scope> - </dependency> - <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-cxf-rest</artifactId> @@ -60,6 +53,13 @@ <artifactId>javax.annotation-api</artifactId> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>${awaitility-version}</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.apache.cxf</groupId> <artifactId>cxf-rt-transports-http-undertow</artifactId> diff --git a/tests/itests-cxfrs/src/test/java/org/apache/camel/kafkaconnector/cxfrs/source/CamelSourceCXFRSITCase.java b/tests/itests-cxfrs/src/test/java/org/apache/camel/kafkaconnector/cxfrs/source/CamelSourceCXFRSITCase.java index 6a4dadf460..2072dc97ca 100644 --- a/tests/itests-cxfrs/src/test/java/org/apache/camel/kafkaconnector/cxfrs/source/CamelSourceCXFRSITCase.java +++ b/tests/itests-cxfrs/src/test/java/org/apache/camel/kafkaconnector/cxfrs/source/CamelSourceCXFRSITCase.java @@ -17,6 +17,8 @@ package org.apache.camel.kafkaconnector.cxfrs.source; +import java.time.Duration; + import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory; import org.apache.camel.kafkaconnector.common.test.CamelSourceTestSupport; import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer; @@ -31,6 +33,7 @@ import org.apache.http.client.methods.HttpGet; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -72,20 +75,16 @@ public class CamelSourceCXFRSITCase extends CamelSourceTestSupport { @Override protected void produceTestData() { TestUtils.waitFor(() -> NetworkUtils.portIsOpen(LOCALHOST, PORT)); - try { - Bus bus = BusFactory.newInstance().createBus(); - - bus.getInInterceptors().add(new LoggingInInterceptor()); - bus.getOutInterceptors().add(new LoggingOutInterceptor()); - try { - doTestGetCustomer("rest"); - } catch (Exception e) { - LOG.info("Test Invocation Failure", e); - } - } catch (Exception e) { - LOG.info("Unable to invoke service: {}", e.getMessage(), e); - fail("Unable to invoke service"); - } + + Bus bus = BusFactory.newInstance().createBus(); + bus.getInInterceptors().add(new LoggingInInterceptor()); + bus.getOutInterceptors().add(new LoggingOutInterceptor()); + + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofSeconds(1)) + .pollDelay(Duration.ZERO) + .untilAsserted(() -> doTestGetCustomer("rest")); } @Override @@ -103,7 +102,7 @@ public class CamelSourceCXFRSITCase extends CamelSourceTestSupport { } @Test - @Timeout(30) + @Timeout(90) public void testBasicSendReceive() { try { String topicName = getTopicForTest(this); diff --git a/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/sink/CamelSinkIdempotentJMSITCase.java b/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/sink/CamelSinkIdempotentJMSITCase.java index 8d36752b56..c028905b0a 100644 --- a/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/sink/CamelSinkIdempotentJMSITCase.java +++ b/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/sink/CamelSinkIdempotentJMSITCase.java @@ -33,8 +33,8 @@ import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient; import org.apache.camel.kafkaconnector.common.test.CamelSinkTestSupport; import org.apache.camel.kafkaconnector.sjms2.clients.JMSClient; import org.apache.camel.kafkaconnector.sjms2.common.SJMS2Common; -import org.apache.camel.test.infra.common.TestUtils; import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.common.TestUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; diff --git a/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/source/CamelSourceJMSWithAggregation.java b/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/source/CamelSourceJMSWithAggregation.java index 9cfa41bb23..9dc5a935b6 100644 --- a/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/source/CamelSourceJMSWithAggregation.java +++ b/tests/itests-sjms2/src/test/java/org/apache/camel/kafkaconnector/sjms2/source/CamelSourceJMSWithAggregation.java @@ -27,8 +27,8 @@ import org.apache.camel.kafkaconnector.common.test.StringMessageConsumer; import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer; import org.apache.camel.kafkaconnector.sjms2.clients.JMSClient; import org.apache.camel.kafkaconnector.sjms2.common.SJMS2Common; -import org.apache.camel.test.infra.common.TestUtils; import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.common.TestUtils; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test;
