You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by oa...@apache.org on 2020/05/06 16:14:22 UTC
[camel] branch CAMEL-14957 created (now fe68084)
This is an automated email from the ASF dual-hosted git repository.
oalsafi pushed a change to branch CAMEL-14957
in repository https://gitbox.apache.org/repos/asf/camel.git.
at fe68084 CAMEL-14957: Fix Kafka testcontainers integration tests
This branch includes the following new commits:
new fe68084 CAMEL-14957: Fix Kafka testcontainers integration tests
The 1 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.
[camel] 01/01: CAMEL-14957: Fix Kafka testcontainers integration
tests
Posted by oa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
oalsafi pushed a commit to branch CAMEL-14957
in repository https://gitbox.apache.org/repos/asf/camel.git
commit fe6808486e6cdec2725da9c33f52f289bb57644e
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Wed May 6 18:13:10 2020 +0200
CAMEL-14957: Fix Kafka testcontainers integration tests
---
.../component/kafka/BaseEmbeddedKafkaTest.java | 24 +++++++--
.../kafka/KafkaConsumerBatchSizeTest.java | 3 ++
.../component/kafka/KafkaConsumerFullTest.java | 3 ++
.../kafka/KafkaConsumerLastRecordHeaderTest.java | 3 ++
.../kafka/KafkaConsumerManualCommitTest.java | 3 ++
.../kafka/KafkaConsumerRebalanceTest.java | 8 +++
.../kafka/KafkaConsumerTopicIsPatternTest.java | 3 ++
.../component/kafka/KafkaProducerFullTest.java | 62 +++++++++++++++-------
8 files changed, 86 insertions(+), 23 deletions(-)
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/BaseEmbeddedKafkaTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/BaseEmbeddedKafkaTest.java
index 4a2c772..3c1f2ef 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/BaseEmbeddedKafkaTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/BaseEmbeddedKafkaTest.java
@@ -20,23 +20,30 @@ import java.util.Properties;
import org.apache.camel.CamelContext;
import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.junit.BeforeClass;
-import org.junit.ClassRule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.wait.strategy.Wait;
-public class BaseEmbeddedKafkaTest extends CamelTestSupport {
+public abstract class BaseEmbeddedKafkaTest extends CamelTestSupport {
+ protected static AdminClient kafkaAdminClient;
+
private static final String CONFLUENT_PLATFORM_VERSION = "5.5.0";
- @ClassRule
- public static KafkaContainer kafkaBroker = new KafkaContainer(CONFLUENT_PLATFORM_VERSION)
+ private static final Logger LOG = LoggerFactory.getLogger(BaseEmbeddedKafkaTest.class);
+
+ protected static KafkaContainer kafkaBroker = new KafkaContainer(CONFLUENT_PLATFORM_VERSION)
.withEmbeddedZookeeper()
.waitingFor(Wait.forListeningPort());
- private static final Logger LOG = LoggerFactory.getLogger(BaseEmbeddedKafkaTest.class);
+ static {
+ kafkaBroker.start();
+ kafkaAdminClient = createAdminClient();
+ }
@BeforeClass
public static void beforeClass() {
@@ -71,4 +78,11 @@ public class BaseEmbeddedKafkaTest extends CamelTestSupport {
protected static String getBootstrapServers() {
return kafkaBroker.getBootstrapServers();
}
+
+ private static AdminClient createAdminClient() {
+ final Properties properties = new Properties();
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker.getBootstrapServers());
+
+ return KafkaAdminClient.create(properties);
+ }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerBatchSizeTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerBatchSizeTest.java
index 06d5de0..4588d8f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerBatchSizeTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerBatchSizeTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Collections;
import java.util.Properties;
import org.apache.camel.Endpoint;
@@ -50,6 +51,8 @@ public class KafkaConsumerBatchSizeTest extends BaseEmbeddedKafkaTest {
if (producer != null) {
producer.close();
}
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
@Override
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
index d6a3150..3673b3f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.kafka;
import java.io.IOException;
+import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.stream.StreamSupport;
@@ -62,6 +63,8 @@ public class KafkaConsumerFullTest extends BaseEmbeddedKafkaTest {
if (producer != null) {
producer.close();
}
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
@Override
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
index c29de33..1cb8673 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -47,6 +48,8 @@ public class KafkaConsumerLastRecordHeaderTest extends BaseEmbeddedKafkaTest {
if (producer != null) {
producer.close();
}
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
/**
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
index 476e346..0e6524d 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.kafka;
import java.io.IOException;
+import java.util.Collections;
import java.util.Properties;
import java.util.stream.StreamSupport;
@@ -55,6 +56,8 @@ public class KafkaConsumerManualCommitTest extends BaseEmbeddedKafkaTest {
if (producer != null) {
producer.close();
}
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
@Override
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerRebalanceTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerRebalanceTest.java
index 856087a..ca19ce0 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerRebalanceTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerRebalanceTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -24,6 +25,7 @@ import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.spi.StateRepository;
+import org.junit.After;
import org.junit.Test;
public class KafkaConsumerRebalanceTest extends BaseEmbeddedKafkaTest {
@@ -48,6 +50,12 @@ public class KafkaConsumerRebalanceTest extends BaseEmbeddedKafkaTest {
assertTrue("StateRepository.getState should have been called twice for topic " + TOPIC + ". Remaining count : " + messagesLatch.getCount(), offsetGetStateCalled);
}
+ @After
+ public void after() {
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
+ }
+
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTopicIsPatternTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTopicIsPatternTest.java
index 9a54c8f..65325aa 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTopicIsPatternTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTopicIsPatternTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Collections;
import java.util.Properties;
import java.util.stream.StreamSupport;
@@ -54,6 +55,8 @@ public class KafkaConsumerTopicIsPatternTest extends BaseEmbeddedKafkaTest {
if (producer != null) {
producer.close();
}
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
@Override
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
index 355c433..19d8b8d 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
@@ -104,24 +104,20 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
@BeforeClass
public static void before() {
- Properties stringsProps = new Properties();
+ stringsConsumerConn = createStringKafkaConsumer("DemoConsumer");
+ bytesConsumerConn = createByteKafkaConsumer(GROUP_BYTES);
+ }
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- stringsConsumerConn = new KafkaConsumer<>(stringsProps);
-
- Properties bytesProps = new Properties();
- bytesProps.putAll(stringsProps);
- bytesProps.put("group.id", GROUP_BYTES);
- bytesProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
- bytesProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
- bytesConsumerConn = new KafkaConsumer<>(bytesProps);
+ @AfterClass
+ public static void after() {
+ // clean all test topics
+ final List<String> topics = new ArrayList<>();
+ topics.add(TOPIC_BYTES);
+ topics.add(TOPIC_INTERCEPTED);
+ topics.add(TOPIC_PROPAGATED_HEADERS);
+ topics.add(TOPIC_STRINGS);
+
+ kafkaAdminClient.deleteTopics(topics);
}
@Override
@@ -329,7 +325,7 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
CountDownLatch messagesLatch = new CountDownLatch(1);
propagatedHeadersTemplate.sendBodyAndHeaders("Some test message", camelHeaders);
- List<ConsumerRecord<String, String>> records = pollForRecords(stringsConsumerConn, TOPIC_PROPAGATED_HEADERS, messagesLatch);
+ List<ConsumerRecord<String, String>> records = pollForRecords(createStringKafkaConsumer("propagatedHeaderConsumer"), TOPIC_PROPAGATED_HEADERS, messagesLatch);
boolean allMessagesReceived = messagesLatch.await(10_000, TimeUnit.MILLISECONDS);
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
@@ -365,6 +361,36 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
return foundHeader.value();
}
+ private static KafkaConsumer<String, String> createStringKafkaConsumer(final String groupId) {
+ Properties stringsProps = new Properties();
+
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ return new KafkaConsumer<>(stringsProps);
+ }
+
+ private static KafkaConsumer<byte[], byte[]> createByteKafkaConsumer(final String groupId) {
+ Properties stringsProps = new Properties();
+
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ return new KafkaConsumer<>(stringsProps);
+ }
+
private List<ConsumerRecord<String, String>> pollForRecords(KafkaConsumer<String, String> consumerConn, String topic, CountDownLatch messagesLatch) {
List<ConsumerRecord<String, String>> consumedRecords = new ArrayList<>();