Merge branch 'main' into add-scylladb

Author: mkorolyov

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -50,6 +50,7 @@ body:
         - Oracle Free
         - Oracle XE
         - OrientDB
+        - Pinecone
         - PostgreSQL
         - Presto
         - Pulsar

.github/ISSUE_TEMPLATE/enhancement.yaml

@@ -50,6 +50,7 @@ body:
         - Oracle Free
         - Oracle XE
         - OrientDB
+        - Pinecone
         - PostgreSQL
         - Presto
         - Pulsar

.github/ISSUE_TEMPLATE/feature.yaml

@@ -50,6 +50,7 @@ body:
         - Oracle Free
         - Oracle XE
         - OrientDB
+        - Pinecone
         - PostgreSQL
         - Qdrant
         - QuestDB

.github/dependabot.yml

@@ -274,6 +274,11 @@ updates:
     schedule:
       interval: "weekly"
     open-pull-requests-limit: 10
+  - package-ecosystem: "gradle"
+    directory: "/modules/pinecone"
+    schedule:
+      interval: "weekly"
+    open-pull-requests-limit: 10
   - package-ecosystem: "gradle"
     directory: "/modules/pulsar"
     schedule:

.github/labeler.yml

@@ -168,6 +168,10 @@
   - changed-files:
     - any-glob-to-any-file:
       - modules/orientdb/**/*
+"modules/pinecone":
+  - changed-files:
+    - any-glob-to-any-file:
+      - modules/pinecone/**/*
 "modules/postgres":
   - changed-files:
     - any-glob-to-any-file:

.github/settings.yml

@@ -211,6 +211,9 @@ labels:
   - name: modules/orientdb
     color: '#006b75'
 
+  - name: modules/pinecone
+    color: '#006b75'
+
   - name: modules/postgres
     color: '#006b75'
 

docs/modules/azure.md

@@ -5,11 +5,14 @@ This module is INCUBATING. While it is ready for use and operational in the curr
 
 Testcontainers module for the Microsoft Azure's [SDK](https://github.com/Azure/azure-sdk-for-java).
 
-Currently, the module supports `Azurite` and `CosmosDB` emulators. In order to use them, you should use the following classes:
+Currently, the module supports `Azurite`, `Azure Event Hubs`, `Azure Service Bus` and `CosmosDB` emulators. In order to use them, you should use the following classes:
 
 Class | Container Image
 -|-
 AzuriteContainer | [mcr.microsoft.com/azure-storage/azurite](https://github.com/microsoft/containerregistry)
+AzureEventHubsContainer | [mcr.microsoft.com/azure-messaging/eventhubs-emulator](https://github.com/microsoft/containerregistry)
+AzureServiceBusEmulatorContainer | [mcr.microsoft.com/azure-messaging/servicebus-emulator](https://github.com/microsoft/containerregistry)
+AzureServiceBusContainer | [mcr.microsoft.com/azure-messaging/servicebus-emulator](https://github.com/microsoft/containerregistry)
 CosmosDBEmulatorContainer | [mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator](https://github.com/microsoft/containerregistry)
 
 ## Usage example
@@ -72,6 +75,66 @@ Build Azure Table client:
 !!! note
     We can use custom credentials the same way as defined in the Blob section.
 
+### Azure Event Hubs Emulator
+
+<!--codeinclude-->
+[Configuring the Azure Event Hubs Emulator container](../../modules/azure/src/test/resources/eventhubs_config.json)
+<!--/codeinclude-->
+
+Start Azure Event Hubs Emulator during a test:
+
+<!--codeinclude-->
+[Setting up a network](../../modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsContainerTest.java) inside_block:network
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Starting an Azurite container as dependency](../../modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsContainerTest.java) inside_block:azuriteContainer
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Starting an Azure Event Hubs Emulator container](../../modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsContainerTest.java) inside_block:emulatorContainer
+<!--/codeinclude-->
+
+#### Using Azure Event Hubs clients
+
+Configure the consumer and the producer clients:
+
+<!--codeinclude-->
+[Configuring the clients](../../modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsContainerTest.java) inside_block:createProducerAndConsumer
+<!--/codeinclude-->
+
+### Azure Service Bus Emulator
+
+<!--codeinclude-->
+[Configuring the Azure Service Bus Emulator container](../../modules/azure/src/test/resources/service-bus-config.json)
+<!--/codeinclude-->
+
+Start Azure Service Bus Emulator during a test:
+
+<!--codeinclude-->
+[Setting up a network](../../modules/azure/src/test/java/org/testcontainers/azure/AzureServiceBusContainerTest.java) inside_block:network
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Starting a SQL Server container as dependency](../../modules/azure/src/test/java/org/testcontainers/azure/AzureServiceBusContainerTest.java) inside_block:sqlContainer
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Starting a Service Bus Emulator container](../../modules/azure/src/test/java/org/testcontainers/azure/AzureServiceBusContainerTest.java) inside_block:emulatorContainer
+<!--/codeinclude-->
+
+#### Using Azure Service Bus clients
+
+Configure the sender and the processor clients:
+
+<!--codeinclude-->
+[Configuring the sender client](../../modules/azure/src/test/java/org/testcontainers/azure/AzureServiceBusContainerTest.java) inside_block:senderClient
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Configuring the processor client](../../modules/azure/src/test/java/org/testcontainers/azure/AzureServiceBusContainerTest.java) inside_block:processorClient
+<!--/codeinclude-->
+
 ### CosmosDB
 
 Start Azure CosmosDB Emulator during a test:

docs/modules/pinecone.md

@@ -0,0 +1,30 @@
+# Pinecone
+
+Testcontainers module for [Pinecone](https://github.com/orgs/pinecone-io/packages/container/package/pinecone-local).
+
+## PineconeContainer's usage examples
+
+You can start an Pinecone container instance from any Java application by using:
+
+<!--codeinclude-->
+[Pinecone container](../../modules/pinecone/src/test/java/org/testcontainers/pinecone/PineconeContainerTest.java) inside_block:container
+<!--/codeinclude-->
+
+## Adding this module to your project dependencies
+
+Add the following dependency to your `pom.xml`/`build.gradle` file:
+
+=== "Gradle"
+```groovy
+testImplementation "org.testcontainers:pinecone:{{latest_version}}"
+```
+
+=== "Maven"
+```xml
+<dependency>
+    <groupId>org.testcontainers</groupId>
+    <artifactId>pinecone</artifactId>
+    <version>{{latest_version}}</version>
+    <scope>test</scope>
+</dependency>
+```

examples/kafka-cluster/src/test/java/com/example/kafkacluster/ApacheKafkaContainerCluster.java

@@ -0,0 +1,104 @@
+package com.example.kafkacluster;
+
+import org.apache.kafka.common.Uuid;
+import org.awaitility.Awaitility;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.kafka.KafkaContainer;
+import org.testcontainers.lifecycle.Startable;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ApacheKafkaContainerCluster implements Startable {
+
+    private final int brokersNum;
+
+    private final Network network;
+
+    private final Collection<KafkaContainer> brokers;
+
+    public ApacheKafkaContainerCluster(String version, int brokersNum, int internalTopicsRf) {
+        if (brokersNum < 0) {
+            throw new IllegalArgumentException("brokersNum '" + brokersNum + "' must be greater than 0");
+        }
+        if (internalTopicsRf < 0 || internalTopicsRf > brokersNum) {
+            throw new IllegalArgumentException(
+                "internalTopicsRf '" + internalTopicsRf + "' must be less than brokersNum and greater than 0"
+            );
+        }
+
+        this.brokersNum = brokersNum;
+        this.network = Network.newNetwork();
+
+        String controllerQuorumVoters = IntStream
+            .range(0, brokersNum)
+            .mapToObj(brokerNum -> String.format("%d@broker-%d:9094", brokerNum, brokerNum))
+            .collect(Collectors.joining(","));
+
+        String clusterId = Uuid.randomUuid().toString();
+
+        this.brokers =
+            IntStream
+                .range(0, brokersNum)
+                .mapToObj(brokerNum -> {
+                    return new KafkaContainer(DockerImageName.parse("apache/kafka").withTag(version))
+                        .withNetwork(this.network)
+                        .withNetworkAliases("broker-" + brokerNum)
+                        .withEnv("CLUSTER_ID", clusterId)
+                        .withEnv("KAFKA_BROKER_ID", brokerNum + "")
+                        .withEnv("KAFKA_NODE_ID", brokerNum + "")
+                        .withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS", controllerQuorumVoters)
+                        .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", internalTopicsRf + "")
+                        .withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")
+                        .withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", internalTopicsRf + "")
+                        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", internalTopicsRf + "")
+                        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", internalTopicsRf + "")
+                        .withStartupTimeout(Duration.ofMinutes(1));
+                })
+                .collect(Collectors.toList());
+    }
+
+    public Collection<KafkaContainer> getBrokers() {
+        return this.brokers;
+    }
+
+    public String getBootstrapServers() {
+        return brokers.stream().map(KafkaContainer::getBootstrapServers).collect(Collectors.joining(","));
+    }
+
+    @Override
+    public void start() {
+        // Needs to start all the brokers at once
+        brokers.parallelStream().forEach(GenericContainer::start);
+
+        Awaitility
+            .await()
+            .atMost(Duration.ofSeconds(30))
+            .untilAsserted(() -> {
+                Container.ExecResult result =
+                    this.brokers.stream()
+                        .findFirst()
+                        .get()
+                        .execInContainer(
+                            "sh",
+                            "-c",
+                            "/opt/kafka/bin/kafka-log-dirs.sh --bootstrap-server localhost:9093 --describe | grep -o '\"broker\"' | wc -l"
+                        );
+                String brokers = result.getStdout().replace("\n", "");
+
+                assertThat(brokers).asInt().isEqualTo(this.brokersNum);
+            });
+    }
+
+    @Override
+    public void stop() {
+        this.brokers.stream().parallel().forEach(GenericContainer::stop);
+    }
+}

examples/kafka-cluster/src/test/java/com/example/kafkacluster/ApacheKafkaContainerClusterTest.java

@@ -0,0 +1,94 @@
+package com.example.kafkacluster;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.tuple;
+
+class ApacheKafkaContainerClusterTest {
+
+    @Test
+    void testKafkaContainerCluster() throws Exception {
+        try (ApacheKafkaContainerCluster cluster = new ApacheKafkaContainerCluster("3.8.0", 3, 2)) {
+            cluster.start();
+            String bootstrapServers = cluster.getBootstrapServers();
+
+            assertThat(cluster.getBrokers()).hasSize(3);
+
+            testKafkaFunctionality(bootstrapServers, 3, 2);
+        }
+    }
+
+    protected void testKafkaFunctionality(String bootstrapServers, int partitions, int rf) throws Exception {
+        try (
+            AdminClient adminClient = AdminClient.create(
+                ImmutableMap.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
+            );
+            KafkaProducer<String, String> producer = new KafkaProducer<>(
+                ImmutableMap.of(
+                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                    bootstrapServers,
+                    ProducerConfig.CLIENT_ID_CONFIG,
+                    UUID.randomUUID().toString()
+                ),
+                new StringSerializer(),
+                new StringSerializer()
+            );
+            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
+                ImmutableMap.of(
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                    bootstrapServers,
+                    ConsumerConfig.GROUP_ID_CONFIG,
+                    "tc-" + UUID.randomUUID(),
+                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+                    "earliest"
+                ),
+                new StringDeserializer(),
+                new StringDeserializer()
+            );
+        ) {
+            String topicName = "messages";
+
+            Collection<NewTopic> topics = Collections.singletonList(new NewTopic(topicName, partitions, (short) rf));
+            adminClient.createTopics(topics).all().get(30, TimeUnit.SECONDS);
+
+            consumer.subscribe(Collections.singletonList(topicName));
+
+            producer.send(new ProducerRecord<>(topicName, "testcontainers", "rulezzz")).get();
+
+            Awaitility
+                .await()
+                .atMost(Duration.ofSeconds(10))
+                .untilAsserted(() -> {
+                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
+
+                    assertThat(records)
+                        .hasSize(1)
+                        .extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
+                        .containsExactly(tuple(topicName, "testcontainers", "rulezzz"));
+                });
+
+            consumer.unsubscribe();
+        }
+    }
+}