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/12/06 18:12:54 UTC
[camel] branch main updated (67c6d52034a -> 10a8482d6b9)
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 67c6d52034a Camel-InfluxDb2: Fixed Endpoint syntax annotation
new 12b643d9614 camel-test-infra-kafka: added a new container for SASL authentication with Kafka
new 10a8482d6b9 (chores) camel-kafka: added an auth-based IT test
The 2 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 +
....java => BaseEmbeddedKafkaAuthTestSupport.java} | 14 ++-
.../kafka/integration/KafkaConsumerAuthIT.java | 139 +++++++++++++++++++++
.../src/test/resources/kafka-jaas.config | 8 ++
.../services/ContainerLocalAuthKafkaService.java | 101 +++++++++++++++
5 files changed, 257 insertions(+), 6 deletions(-)
copy components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/{BaseEmbeddedKafkaTestSupport.java => BaseEmbeddedKafkaAuthTestSupport.java} (87%)
create mode 100644 components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
create mode 100644 components/camel-kafka/src/test/resources/kafka-jaas.config
create mode 100644 test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalAuthKafkaService.java
[camel] 02/02: (chores) camel-kafka: added an auth-based IT test
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 10a8482d6b967bacdaba80795b817d386075d9df
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Tue Dec 6 15:24:47 2022 +0100
(chores) camel-kafka: added an auth-based IT test
Pre-work for verifying/fixing CAMEL-18796
---
components/camel-kafka/pom.xml | 1 +
.../BaseEmbeddedKafkaAuthTestSupport.java | 103 +++++++++++++++
.../kafka/integration/KafkaConsumerAuthIT.java | 139 +++++++++++++++++++++
.../src/test/resources/kafka-jaas.config | 8 ++
4 files changed, 251 insertions(+)
diff --git a/components/camel-kafka/pom.xml b/components/camel-kafka/pom.xml
index 186b184888f..50a47d63950 100644
--- a/components/camel-kafka/pom.xml
+++ b/components/camel-kafka/pom.xml
@@ -173,6 +173,7 @@
</systemPropertyVariables>
<reuseForks>true</reuseForks>
<reportNameSuffix>kafka-2</reportNameSuffix>
+ <excludedGroups>non-abstract</excludedGroups>
</configuration>
<goals>
<goal>integration-test</goal>
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java
new file mode 100644
index 00000000000..6e7f11f8c96
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java
@@ -0,0 +1,103 @@
+/*
+ * 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.component.kafka.integration;
+
+import java.util.Properties;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaService;
+import org.apache.camel.test.junit5.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.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Tags;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Tags({ @Tag("non-abstract") })
+public abstract class BaseEmbeddedKafkaAuthTestSupport extends CamelTestSupport {
+ @RegisterExtension
+ public static ContainerLocalAuthKafkaService service = new ContainerLocalAuthKafkaService("/kafka-jaas.config");
+
+ protected static AdminClient kafkaAdminClient;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseEmbeddedKafkaAuthTestSupport.class);
+
+ @BeforeAll
+ public static void beforeClass() {
+ LOG.info("Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
+ System.setProperty("bootstrapServers", service.getBootstrapServers());
+ }
+
+ @BeforeEach
+ public void setKafkaAdminClient() {
+ if (kafkaAdminClient == null) {
+ kafkaAdminClient = createAdminClient();
+ }
+ }
+
+ public static Properties getDefaultProperties(KafkaService service) {
+ LOG.info("Connecting to Kafka {}", service.getBootstrapServers());
+
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, service.getBootstrapServers());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_SERIALIZER);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_SERIALIZER);
+ props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_PARTITIONER);
+ props.put(ProducerConfig.ACKS_CONFIG, "1");
+ return props;
+ }
+
+ protected Properties getDefaultProperties() {
+ return getDefaultProperties(service);
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getPropertiesComponent().setLocation("ref:prop");
+
+ KafkaComponent kafka = new KafkaComponent(context);
+ kafka.init();
+ kafka.getConfiguration().setBrokers(service.getBootstrapServers());
+ context.addComponent("kafka", kafka);
+
+ return context;
+ }
+
+ protected static String getBootstrapServers() {
+ return service.getBootstrapServers();
+ }
+
+ private static AdminClient createAdminClient() {
+ return createAdminClient(service);
+ }
+
+ public static AdminClient createAdminClient(KafkaService service) {
+ final Properties properties = new Properties();
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, service.getBootstrapServers());
+
+ return KafkaAdminClient.create(properties);
+ }
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
new file mode 100644
index 00000000000..068e4ebeda1
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
@@ -0,0 +1,139 @@
+/*
+ * 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.component.kafka.integration;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.StreamSupport;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
+ public static final String TOPIC = "test-auth-full";
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthIT.class);
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+
+ @BeforeEach
+ public void before() {
+ Properties props = getDefaultProperties();
+ props.put("sasl.jaas.config", ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret"));
+ props.put("security.protocol", "SASL_PLAINTEXT");
+ props.put("sasl.mechanism", "PLAIN");
+
+ try {
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+
+ MockConsumerInterceptor.recordsCaptured.clear();
+ }
+
+ @AfterEach
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+ final String simpleSaslJaasConfig
+ = ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret");
+
+ fromF("kafka:%s"
+ + "?groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=%s"
+ + "&saslMechanism=PLAIN&securityProtocol=SASL_PLAINTEXT&saslJaasConfig=%s", TOPIC,
+ "KafkaConsumerAuthIT", "org.apache.camel.component.kafka.MockConsumerInterceptor", simpleSaslJaasConfig)
+ .process(
+ exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
+ .routeId("full-it").to(to);
+ }
+ };
+ }
+
+ @Timeout(30)
+ @Order(3)
+ @Test
+ public void kafkaMessageIsConsumedByCamel() throws InterruptedException {
+ String propagatedHeaderKey = "PropagatedCustomHeader";
+ byte[] propagatedHeaderValue = "propagated header value".getBytes();
+ String skippedHeaderKey = "CamelSkippedHeader";
+ to.expectedMessageCount(5);
+ to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
+ // The LAST_RECORD_BEFORE_COMMIT header should not be configured on any
+ // exchange because autoCommitEnable=true
+ to.expectedHeaderValuesReceivedInAnyOrder(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, null, null, null, null, null);
+ to.expectedHeaderReceived(propagatedHeaderKey, propagatedHeaderValue);
+
+ for (int k = 0; k < 5; k++) {
+ String msg = "message-" + k;
+ ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
+ data.headers().add(new RecordHeader("CamelSkippedHeader", "skipped header value".getBytes()));
+ data.headers().add(new RecordHeader(propagatedHeaderKey, propagatedHeaderValue));
+ producer.send(data);
+ }
+
+ to.assertIsSatisfied(3000);
+
+ 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");
+ assertTrue(headers.containsKey(propagatedHeaderKey), "Should receive propagated header");
+ }
+
+}
diff --git a/components/camel-kafka/src/test/resources/kafka-jaas.config b/components/camel-kafka/src/test/resources/kafka-jaas.config
new file mode 100644
index 00000000000..eb54850d5d1
--- /dev/null
+++ b/components/camel-kafka/src/test/resources/kafka-jaas.config
@@ -0,0 +1,8 @@
+KafkaServer {
+ org.apache.kafka.common.security.plain.PlainLoginModule required
+ serviceName="kafka"
+ username="admin"
+ password="admin-secret"
+ user_admin="admin-secret"
+ user_camel="camel-secret";
+};
[camel] 01/02: camel-test-infra-kafka: added a new container for SASL authentication with Kafka
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 12b643d9614c38e9ca456d4b4e7bf9fd735e52b7
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Tue Dec 6 15:23:44 2022 +0100
camel-test-infra-kafka: added a new container for SASL authentication with Kafka
---
.../services/ContainerLocalAuthKafkaService.java | 101 +++++++++++++++++++++
1 file changed, 101 insertions(+)
diff --git a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalAuthKafkaService.java b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalAuthKafkaService.java
new file mode 100644
index 00000000000..17003028747
--- /dev/null
+++ b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/ContainerLocalAuthKafkaService.java
@@ -0,0 +1,101 @@
+/*
+ * 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.apache.camel.test.infra.common.services.ContainerService;
+import org.apache.camel.test.infra.kafka.common.KafkaProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
+
+/**
+ * A KafkaContainer that supports JAAS+SASL based authentication
+ */
+public class ContainerLocalAuthKafkaService implements KafkaService, ContainerService<KafkaContainer> {
+ private static final Logger LOG = LoggerFactory.getLogger(ContainerLocalAuthKafkaService.class);
+ private final KafkaContainer kafka;
+ private final String jaasConfigFile;
+
+ public ContainerLocalAuthKafkaService(String jaasConfigFile) {
+ this.jaasConfigFile = jaasConfigFile;
+ kafka = initContainer();
+ }
+
+ public ContainerLocalAuthKafkaService(KafkaContainer kafka, String jaasConfigFile) {
+ this.kafka = kafka;
+ this.jaasConfigFile = jaasConfigFile;
+ }
+
+ protected KafkaContainer initContainer() {
+ final KafkaContainer container = new KafkaContainer(DockerImageName.parse(ContainerLocalKafkaService.KAFKA3_IMAGE_NAME))
+ .withEmbeddedZookeeper();
+
+ final MountableFile mountableFile = MountableFile.forClasspathResource(jaasConfigFile);
+ LOG.debug("Using mountable file at: {}", mountableFile.getFilesystemPath());
+ container.withCopyFileToContainer(mountableFile, "/tmp/kafka-jaas.config");
+
+ container.withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=/tmp/kafka-jaas.config")
+ .withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092")
+ .withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN")
+ .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:SASL_PLAINTEXT,BROKER:PLAINTEXT")
+ .withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
+
+ return container;
+ }
+
+ public String getBootstrapServers() {
+ return kafka.getBootstrapServers();
+ }
+
+ @Override
+ public void registerProperties() {
+ System.setProperty(KafkaProperties.KAFKA_BOOTSTRAP_SERVERS, getBootstrapServers());
+ }
+
+ @Override
+ public void initialize() {
+ kafka.start();
+ registerProperties();
+
+ LOG.info("Kafka bootstrap server running at address {}", kafka.getBootstrapServers());
+ }
+
+ @Override
+ public void shutdown() {
+ kafka.stop();
+ }
+
+ @Override
+ public KafkaContainer getContainer() {
+ return kafka;
+ }
+
+ /**
+ * This method can be used by tests to get a sample 'sasl.jaas.config' configuration for the given user and password
+ *
+ * @param username the user to create the config for
+ * @param password the password for the user
+ * @return A string with the configuration
+ */
+ public static String generateSimpleSaslJaasConfig(String username, String password) {
+ return String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=%s password=%s;",
+ username, password);
+ }
+}