From 0bb6925fb7d420dbd2c5d86f2bbf1d8513fd7dfc Mon Sep 17 00:00:00 2001 From: gustavomonarin Date: Thu, 27 Oct 2022 19:04:21 +0200 Subject: [PATCH] Expose Redpanda schema registry (#5994) The schema registry port `8081` was added to the list of exposed ports and a supporting method `getSchemaRegistryAddress` was added to easily use for configuring `schema.registry.url` serializer configuration. A minor change on the internal advertised address was made as the `kafka` alias was not available within the container and the panda proxy was getting lost with the following message: ``` INFO 2022-10-17 19:08:37,164 [shard 0] kafka/client - broker.cc:41 - connected to broker:-1 - 0.0.0.0:29092 WARN 2022-10-17 19:08:37,176 [shard 0] kafka/client - broker.cc:52 - std::system_error: kafka: Not found ERROR 2022-10-17 19:08:37,177 [shard 0] pandaproxy - service.cc:137 - Schema registry failed to initialize internal topic: kafka::client::broker_error ({ node: -1 }, { error_code: broker_not_available [8] }) ``` --- docs/modules/redpanda.md | 6 ++++ modules/redpanda/build.gradle | 1 + .../redpanda/RedpandaContainer.java | 11 +++++-- .../redpanda/RedpandaContainerTest.java | 32 +++++++++++++++++++ 4 files changed, 48 insertions(+), 2 deletions(-) diff --git a/docs/modules/redpanda.md b/docs/modules/redpanda.md index 8749c9c87f5..8cdcc1ee83d 100644 --- a/docs/modules/redpanda.md +++ b/docs/modules/redpanda.md @@ -19,6 +19,12 @@ Now your tests or any other process running on your machine can get access to ru [Bootstrap Servers](../../modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java) inside_block:getBootstrapServers +Redpanda also provides a schema registry implementation. Like the Redpanda broker, you can access by using the following schema registry location: + + +[Schema Registry](../../modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java) inside_block:getSchemaRegistryAddress + + ## Adding this module to your project dependencies Add the following dependency to your `pom.xml`/`build.gradle` file: diff --git a/modules/redpanda/build.gradle b/modules/redpanda/build.gradle index 4d7221711a3..17e0ca201bc 100644 --- a/modules/redpanda/build.gradle +++ b/modules/redpanda/build.gradle @@ -5,4 +5,5 @@ dependencies { testImplementation 'org.apache.kafka:kafka-clients:3.3.0' testImplementation 'org.assertj:assertj-core:3.23.1' + testImplementation 'io.rest-assured:rest-assured:5.2.0' } diff --git a/modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java b/modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java index cbb7db1e7c1..32b2ae20857 100644 --- a/modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java +++ b/modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java @@ -18,6 +18,8 @@ public class RedpandaContainer extends GenericContainer { private static final int REDPANDA_PORT = 9092; + private static final int SCHEMA_REGISTRY_PORT = 8081; + private static final String STARTER_SCRIPT = "/testcontainers_start.sh"; public RedpandaContainer(String image) { @@ -33,7 +35,7 @@ public RedpandaContainer(DockerImageName imageName) { throw new IllegalArgumentException("Redpanda version must be >= v22.2.1"); } - withExposedPorts(REDPANDA_PORT); + withExposedPorts(REDPANDA_PORT, SCHEMA_REGISTRY_PORT); withCreateContainerCmdModifier(cmd -> { cmd.withEntrypoint("sh"); }); @@ -49,7 +51,8 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) { command += "/usr/bin/rpk redpanda start --mode dev-container "; command += "--kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092 "; - command += "--advertise-kafka-addr PLAINTEXT://kafka:29092,OUTSIDE://" + getHost() + ":" + getMappedPort(9092); + command += + "--advertise-kafka-addr PLAINTEXT://127.0.0.1:29092,OUTSIDE://" + getHost() + ":" + getMappedPort(9092); copyFileToContainer(Transferable.of(command, 0777), STARTER_SCRIPT); } @@ -57,4 +60,8 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) { public String getBootstrapServers() { return String.format("PLAINTEXT://%s:%s", getHost(), getMappedPort(REDPANDA_PORT)); } + + public String getSchemaRegistryAddress() { + return String.format("http://%s:%s", getHost(), getMappedPort(SCHEMA_REGISTRY_PORT)); + } } diff --git a/modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java b/modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java index 086bf43364b..5ffccd63c62 100644 --- a/modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java +++ b/modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java @@ -1,6 +1,8 @@ package org.testcontainers.redpanda; import com.google.common.collect.ImmutableMap; +import io.restassured.RestAssured; +import io.restassured.response.Response; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.NewTopic; @@ -64,6 +66,36 @@ public void testNotCompatibleVersion() { .hasMessageContaining("Redpanda version must be >= v22.2.1"); } + @Test + public void testSchemaRegistry() { + try (RedpandaContainer container = new RedpandaContainer(REDPANDA_DOCKER_IMAGE)) { + container.start(); + + String subjectsEndpoint = String.format( + "%s/subjects", + // getSchemaRegistryAddress { + container.getSchemaRegistryAddress() + // } + ); + + String subjectName = String.format("test-%s-value", UUID.randomUUID()); + + Response createSubject = RestAssured + .given() + .contentType("application/vnd.schemaregistry.v1+json") + .pathParam("subject", subjectName) + .body("{\"schema\": \"{\\\"type\\\": \\\"string\\\"}\"}") + .when() + .post(subjectsEndpoint + "/{subject}/versions") + .thenReturn(); + assertThat(createSubject.getStatusCode()).isEqualTo(200); + + Response allSubjects = RestAssured.given().when().get(subjectsEndpoint).thenReturn(); + assertThat(allSubjects.getStatusCode()).isEqualTo(200); + assertThat(allSubjects.jsonPath().getList("$")).contains(subjectName); + } + } + private void testKafkaFunctionality(String bootstrapServers) throws Exception { testKafkaFunctionality(bootstrapServers, 1, 1); }