You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2022/03/24 10:25:00 UTC
[camel] branch main updated (c68d593 -> c7a7f4f)
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git.
from c68d593 CAMEL-17615: filter unused options in the Kafka client to prevent warnings
new 8e3ee21 [CAMEL-17845] verify integration tests results
new 2e97841 [CAMEL-17845] fix flaky tests
new dacf6e6 [CAMEL-17845] Use single Kafka server when running IT
new c7a7f4f [CAMEL-17845][CAMEL-17725] Make singleton services thread safe
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
components/camel-kafka/pom.xml | 1 +
.../integration/BaseEmbeddedKafkaTestSupport.java | 4 ++-
.../kafka/integration/KafkaConsumerFullIT.java | 4 +--
.../integration/KafkaConsumerHealthCheckIT.java | 4 +--
.../integration/KafkaConsumerManualCommitIT.java | 2 ++
.../integration/KafkaConsumerTopicIsPatternIT.java | 5 ++--
.../FhirLocalSingletonContainerService.java | 11 +++-----
.../ContainerLocalSingletonKafkaService.java} | 30 +++++++++++++++-------
.../infra/kafka/services/KafkaServiceFactory.java | 9 +++++++
9 files changed, 45 insertions(+), 25 deletions(-)
copy test-infra/{camel-test-infra-fhir/src/test/java/org/apache/camel/test/infra/fhir/services/FhirLocalSingletonContainerService.java => camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java} (58%)
[camel] 01/04: [CAMEL-17845] verify integration tests results
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8e3ee21024609ed37782c687a8f74a3a2247d6d5
Author: John Poth <po...@gmail.com>
AuthorDate: Wed Mar 23 13:07:29 2022 +0100
[CAMEL-17845] verify integration tests results
---
components/camel-kafka/pom.xml | 1 +
1 file changed, 1 insertion(+)
diff --git a/components/camel-kafka/pom.xml b/components/camel-kafka/pom.xml
index 7bd0134..7fe17b4 100644
--- a/components/camel-kafka/pom.xml
+++ b/components/camel-kafka/pom.xml
@@ -119,6 +119,7 @@
</configuration>
<goals>
<goal>integration-test</goal>
+ <goal>verify</goal>
</goals>
</execution>
</executions>
[camel] 02/04: [CAMEL-17845] fix flaky tests
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 2e97841cd5e9a70c3072c73f16ceef600a887481
Author: John Poth <po...@gmail.com>
AuthorDate: Wed Mar 23 13:08:28 2022 +0100
[CAMEL-17845] fix flaky tests
---
.../camel/component/kafka/integration/KafkaConsumerFullIT.java | 4 ++--
.../component/kafka/integration/KafkaConsumerHealthCheckIT.java | 4 ++--
.../component/kafka/integration/KafkaConsumerManualCommitIT.java | 2 ++
.../component/kafka/integration/KafkaConsumerTopicIsPatternIT.java | 5 ++---
4 files changed, 8 insertions(+), 7 deletions(-)
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
index 2e0792e..6a3d4af 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
@@ -117,8 +117,8 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
to.assertIsSatisfied(3000);
- assertEquals(5, StreamSupport.stream(MockConsumerInterceptor.recordsCaptured.get(0).records(TOPIC).spliterator(), false)
- .count());
+ assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
assertFalse(headers.containsKey(skippedHeaderKey), "Should not receive skipped header");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
index 6295361..576e011 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
@@ -155,8 +155,8 @@ public class KafkaConsumerHealthCheckIT extends BaseEmbeddedKafkaTestSupport {
to.assertIsSatisfied(3000);
- assertEquals(5, StreamSupport.stream(MockConsumerInterceptor.recordsCaptured.get(0).records(TOPIC).spliterator(), false)
- .count());
+ assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
assertFalse(headers.containsKey(skippedHeaderKey), "Should not receive skipped header");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
index e57e25f..e92a55e 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
@@ -29,9 +29,11 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.TestInstance;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+@TestInstance(TestInstance.Lifecycle.PER_METHOD)
public class KafkaConsumerManualCommitIT extends BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "testManualCommitTest";
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
index ebe39dc..bee7630 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
@@ -89,11 +89,10 @@ public class KafkaConsumerTopicIsPatternIT extends BaseEmbeddedKafkaTestSupport
ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
producer.send(data);
}
-
to.assertIsSatisfied(3000);
- assertEquals(5, StreamSupport.stream(MockConsumerInterceptor.recordsCaptured.get(0).records(TOPIC).spliterator(), false)
- .count());
+ assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
}
}
[camel] 03/04: [CAMEL-17845] Use single Kafka server when running IT
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit dacf6e603ba4883a56e0553566ec48ea6a16fdf5
Author: John Poth <po...@gmail.com>
AuthorDate: Wed Mar 23 13:09:25 2022 +0100
[CAMEL-17845] Use single Kafka server when running IT
---
.../integration/BaseEmbeddedKafkaTestSupport.java | 4 +-
.../ContainerLocalSingletonKafkaService.java | 61 ++++++++++++++++++++++
.../infra/kafka/services/KafkaServiceFactory.java | 9 ++++
3 files changed, 73 insertions(+), 1 deletion(-)
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
index 10d54c1..ff427f2 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
@@ -29,14 +29,16 @@ import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public abstract class BaseEmbeddedKafkaTestSupport extends CamelTestSupport {
// Note: you must use Strimzi on MacOS with Apple Silicon (and possibly other ARM-based architectures)
@RegisterExtension
- public static KafkaService service = KafkaServiceFactory.createService();
+ public static KafkaService service = KafkaServiceFactory.createSingletonService();
protected static AdminClient kafkaAdminClient;
diff --git a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java
new file mode 100644
index 0000000..0e5d974
--- /dev/null
+++ b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.kafka.services;
+
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+public class ContainerLocalSingletonKafkaService extends ContainerLocalKafkaService
+ implements ExtensionContext.Store.CloseableResource {
+ private static boolean started;
+
+ public ContainerLocalSingletonKafkaService(KafkaContainer container) {
+ super(container);
+ }
+
+ public ContainerLocalSingletonKafkaService() {
+ super();
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext extensionContext) {
+ if (!started) {
+ started = true;
+ extensionContext.getRoot().getStore(ExtensionContext.Namespace.GLOBAL).put("kafka", this);
+ super.initialize();
+ }
+ }
+
+ @Override
+ public void afterAll(ExtensionContext extensionContext) {
+ // no op
+ }
+
+ @Override
+ public void close() {
+ super.shutdown();
+ }
+
+ public static ContainerLocalSingletonKafkaService kafka3Container() {
+ KafkaContainer container = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.0.1"));
+ container = container.withEmbeddedZookeeper();
+
+ return new ContainerLocalSingletonKafkaService(container);
+ }
+}
diff --git a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaServiceFactory.java b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaServiceFactory.java
index ea237d6..43ba6b3 100644
--- a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaServiceFactory.java
+++ b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaServiceFactory.java
@@ -36,4 +36,13 @@ public final class KafkaServiceFactory {
.addMapping("local-kafka3-container", ContainerLocalKafkaService::kafka3Container)
.build();
}
+
+ public static KafkaService createSingletonService() {
+ return builder()
+ .addLocalMapping(() -> new ContainerLocalSingletonKafkaService())
+ .addRemoteMapping(RemoteKafkaService::new)
+ .addMapping("local-kafka3-container", ContainerLocalSingletonKafkaService::kafka3Container)
+ .build();
+ }
+
}
[camel] 04/04: [CAMEL-17845][CAMEL-17725] Make singleton services thread safe
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c7a7f4f346036cf5356c751099c186b8bb846414
Author: John Poth <po...@gmail.com>
AuthorDate: Wed Mar 23 15:12:06 2022 +0100
[CAMEL-17845][CAMEL-17725] Make singleton services thread safe
---
.../fhir/services/FhirLocalSingletonContainerService.java | 11 +++--------
.../kafka/services/ContainerLocalSingletonKafkaService.java | 13 +++++--------
2 files changed, 8 insertions(+), 16 deletions(-)
diff --git a/test-infra/camel-test-infra-fhir/src/test/java/org/apache/camel/test/infra/fhir/services/FhirLocalSingletonContainerService.java b/test-infra/camel-test-infra-fhir/src/test/java/org/apache/camel/test/infra/fhir/services/FhirLocalSingletonContainerService.java
index 6860473..41f9e2e 100644
--- a/test-infra/camel-test-infra-fhir/src/test/java/org/apache/camel/test/infra/fhir/services/FhirLocalSingletonContainerService.java
+++ b/test-infra/camel-test-infra-fhir/src/test/java/org/apache/camel/test/infra/fhir/services/FhirLocalSingletonContainerService.java
@@ -21,17 +21,12 @@ import org.junit.jupiter.api.extension.ExtensionContext;
public class FhirLocalSingletonContainerService extends FhirLocalContainerService
implements ExtensionContext.Store.CloseableResource {
- private static boolean started;
-
@Override
public void beforeAll(ExtensionContext extensionContext) {
- if (!started) {
- started = true;
- // Your "before all tests" startup logic goes here
- // The following line registers a callback hook when the root test context is shut down
- extensionContext.getRoot().getStore(ExtensionContext.Namespace.GLOBAL).put("fhir", this);
+ extensionContext.getRoot().getStore(ExtensionContext.Namespace.GLOBAL).getOrComputeIfAbsent("fhir", s -> {
super.initialize();
- }
+ return this;
+ });
}
@Override
diff --git a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java
index 0e5d974..3f27568 100644
--- a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java
+++ b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalSingletonKafkaService.java
@@ -23,7 +23,6 @@ import org.testcontainers.utility.DockerImageName;
public class ContainerLocalSingletonKafkaService extends ContainerLocalKafkaService
implements ExtensionContext.Store.CloseableResource {
- private static boolean started;
public ContainerLocalSingletonKafkaService(KafkaContainer container) {
super(container);
@@ -35,11 +34,10 @@ public class ContainerLocalSingletonKafkaService extends ContainerLocalKafkaServ
@Override
public void beforeAll(ExtensionContext extensionContext) {
- if (!started) {
- started = true;
- extensionContext.getRoot().getStore(ExtensionContext.Namespace.GLOBAL).put("kafka", this);
+ extensionContext.getRoot().getStore(ExtensionContext.Namespace.GLOBAL).getOrComputeIfAbsent("kafka", s -> {
super.initialize();
- }
+ return this;
+ });
}
@Override
@@ -53,9 +51,8 @@ public class ContainerLocalSingletonKafkaService extends ContainerLocalKafkaServ
}
public static ContainerLocalSingletonKafkaService kafka3Container() {
- KafkaContainer container = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.0.1"));
- container = container.withEmbeddedZookeeper();
-
+ KafkaContainer container
+ = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.0.1")).withEmbeddedZookeeper();
return new ContainerLocalSingletonKafkaService(container);
}
}