Skip to content

Commit

Permalink
wip final extension manager ITest
Browse files Browse the repository at this point in the history
  • Loading branch information
@pj-spoelders
pj-spoelders committed Sep 27, 2023
1 parent a58229a commit 4c85ff4
Show file tree
Hide file tree
Showing 3 changed files with 66 additions and 23 deletions.
4 changes: 4 additions & 0 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -380,6 +380,10 @@
<!-- Excludes netty-handler and aXMLRPC vulnerabilities, both have no fix as of 11.09.23 -->
<exclude>CVE-2023-4586</exclude>
<exclude>CVE-2020-36641</exclude>
<exclude>CVE-2023-40167</exclude>
<exclude>CVE-2023-36479</exclude>
<exclude>CVE-2023-43642</exclude>

</excludeVulnerabilityIds>
</configuration>
</plugin>
Expand Down
3 changes: 2 additions & 1 deletion src/test/java/com/exasol/cloudetl/kafka/ExtensionIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -281,7 +281,8 @@ private void executeKafkaImport(final Table targetTable, final KafkaTestSetup ka
// topics. Avro is set as default record value format.
" TOPIC_NAME = '" + kafkaSetup.getTopicName() + "'\n" + //
" TABLE_NAME = '" + targetTable.getFullyQualifiedName() + "'\n" + //
" GROUP_ID = 'exasol-kafka-udf-consumers'\n";
" GROUP_ID = 'exaudf' \n" + //
" CONSUME_ALL_OFFSETS = 'true' \n";
LOGGER.info("Executing query '" + sql + "'");
executeStatement(sql);
}
Expand Down
82 changes: 60 additions & 22 deletions src/test/java/com/exasol/cloudetl/kafka/KafkaTestSetup.java
View file
Original file line number Diff line number Diff line change
@@ -1,13 +1,14 @@
package com.exasol.cloudetl.kafka;

import java.util.Collections;
import java.util.Properties;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.logging.Logger;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.KafkaFuture;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.utility.DockerImageName;

class KafkaTestSetup implements AutoCloseable {
Expand All @@ -31,17 +32,47 @@ public String getTopicName() {
return this.topicName;
}

private static final int ZOOKEEPER_PORT = 2181;
private static final int KAFKA_EXTERNAL_PORT = 29092;
private static final int SCHEMA_REGISTRY_PORT = 8081;
private static final int ADMIN_TIMEOUT_MILLIS = 5000;

static KafkaTestSetup create() throws ExecutionException, InterruptedException {

final DockerImageName dockerImageName = DockerImageName.parse("confluentinc/cp-kafka:latest");
@SuppressWarnings("resource")
final GenericContainer<?> zookeeperContainer = new GenericContainer<>("confluentinc/cp-zookeeper:7.4.1")
// .withNetwork(kafka.getNetwork())
// .withNetworkAliases("zookeeper")
// .withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
.withExposedPorts(ZOOKEEPER_PORT)// ;
.withEnv("ZOOKEEPER_CLIENT_PORT", Integer.toString(ZOOKEEPER_PORT));
// c.withReuse(true)
zookeeperContainer.start();

final String zookeeperConnString = "172.17.0.1" + ":" + zookeeperContainer.getMappedPort(ZOOKEEPER_PORT);
@SuppressWarnings("resource") // Container will be stopped in close() method
final DockerImageName dockerImageName = DockerImageName.parse("confluentinc/cp-kafka:6.2.1");
final org.testcontainers.containers.KafkaContainer container = new org.testcontainers.containers.KafkaContainer(
dockerImageName);
container.start();
final org.testcontainers.containers.KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.4.1"))//
.withExternalZookeeper(zookeeperConnString)
// .withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:$ZOOKEEPER_PORT")//
.withEnv("KAFKA_BROKER_ID", "0")//
// .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT")//
// .withEnv("KAFKA_LISTENERS", "INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:" + KAFKA_EXTERNAL_PORT)
// .withEnv("KAFKA_ADVERTISED_LISTENERS",
// "INTERNAL://kafka01:9092,EXTERNAL://127.0.0.1:" + KAFKA_EXTERNAL_PORT)
// .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "INTERNAL")//
.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")//
.withExtraHost("kafka01", "127.0.0.1")//
.dependsOn(zookeeperContainer);// .withEmbeddedZookeeper();//

LOG.info("zookeeper: " + zookeeperConnString); // .withKraft();
kafkaContainer.start();
// return String.format("PLAINTEXT://%s:%s", this.getHost(), this.getMappedPort(9093));
// container.start();

final Properties properties = new Properties();
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, container.getBootstrapServers());
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
final String topicName = "testTopic";
LOG.info("Topicname: " + topicName);
// CREATE TOPIC AND WAIT
Expand All @@ -53,32 +84,39 @@ static KafkaTestSetup create() throws ExecutionException, InterruptedException {
final CreateTopicsResult result = admin.createTopics(Collections.singleton(newTopic));

final KafkaFuture<Void> future = result.values().get(topicName);
final var createTopicsResult = future.get();
future.get();

final Set<String> topicNames = admin.listTopics().names().get();

LOG.info("Succesfully created topic");
} catch (final Exception ex) {
LOG.warning("Exception occurred during Kafka topic creation: '" + ex.getMessage() + "'");
}
// PRODUCE
final Properties producerProps = new Properties();
producerProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
producerProps.put("acks", "all");
producerProps.put("retries", 0);
producerProps.put("batch.size", 16384);
producerProps.put("linger.ms", 1);
producerProps.put("buffer.memory", 33554432);
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

final Producer<String, String> producer = new KafkaProducer<>(producerProps);
try {
// PRODUCE
final Properties producerProps = new Properties();
producerProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, container.getBootstrapServers());
producerProps.put("acks", "all");
producerProps.put("retries", 0);
producerProps.put("batch.size", 16384);
producerProps.put("linger.ms", 1);
producerProps.put("buffer.memory", 33554432);
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

final Producer<String, String> producer = new KafkaProducer<>(producerProps);

// for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(1), "OK"));
producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(2), "WARN"));
producer.close();

} catch (final Exception ex) {
LOG.warning("Exception occurred producing Kafka records: '" + ex.getMessage() + "'");
} finally {
producer.close();
}
// RETURN kafkaTestSetup object
return new KafkaTestSetup(container, topicName);
return new KafkaTestSetup(kafkaContainer, topicName);
}

@Override
Expand Down

0 comments on commit 4c85ff4

Please sign in to comment.