diff --git a/examples/kafka/src/test/resources/strimzi-custom-server-ssl.properties b/examples/kafka/src/test/resources/strimzi-custom-server-ssl.properties index cdad609b6..aaa5f8495 100644 --- a/examples/kafka/src/test/resources/strimzi-custom-server-ssl.properties +++ b/examples/kafka/src/test/resources/strimzi-custom-server-ssl.properties @@ -17,8 +17,14 @@ ############################# Server Basics ############################# -# The id of the broker. This must be set to a unique integer for each broker. -broker.id=0 +# The role of this server. Setting this puts us in KRaft mode +process.roles=broker,controller + +# The node id associated with this instance's roles +node.id=1 + +# The connect string for the controller quorum +controller.quorum.voters=1@localhost:9094 ############################# Socket Server Settings ############################# @@ -29,19 +35,25 @@ broker.id=0 # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092 -listeners=BROKER://0.0.0.0:9093,SSL://0.0.0.0:9092 - +listeners=BROKER://0.0.0.0:9093,SSL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094 +# Name of listener used for communication between brokers. +inter.broker.listener.name=BROKER # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 -advertised.listeners=SSL://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 +advertised.listeners=SSL://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 + +# A comma-separated list of the names of the listeners used by the controller. +# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol +# This is required if running in KRaft mode. +controller.listener.names=CONTROLLER # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL -listener.security.protocol.map=BROKER:PLAINTEXT,SSL:SSL +listener.security.protocol.map=BROKER:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 @@ -58,10 +70,7 @@ socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 -inter.broker.listener.name=BROKER - #### SSL #### - ssl.keystore.location=/opt/kafka/config/strimzi-custom-server-ssl-keystore.p12 ssl.keystore.password=top-secret ssl.keystore.type=PKCS12 @@ -75,7 +84,7 @@ ssl.endpoint.identification.algorithm= ############################# Log Basics ############################# # A comma separated list of directories under which to store log files -log.dirs=/tmp/kafka-logs +log.dirs=/tmp/kraft-combined-logs # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across @@ -130,25 +139,3 @@ log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 - -############################# Zookeeper ############################# - -# Zookeeper connection string (see zookeeper docs for details). -# This is a comma separated host:port pairs, each corresponding to a zk -# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". -# You can also append an optional chroot string to the urls to specify the -# root directory for all kafka znodes. -zookeeper.connect=localhost:2181 - -# Timeout in ms for connecting to zookeeper -zookeeper.connection.timeout.ms=45000 - - -############################# Group Coordinator Settings ############################# - -# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. -# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. -# The default value for this is 3 seconds. -# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. -# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. -group.initial.rebalance.delay.ms=0 diff --git a/quarkus-test-containers/src/main/java/io/quarkus/test/services/containers/DockerContainerManagedResource.java b/quarkus-test-containers/src/main/java/io/quarkus/test/services/containers/DockerContainerManagedResource.java index fe0fb006c..c5792e289 100644 --- a/quarkus-test-containers/src/main/java/io/quarkus/test/services/containers/DockerContainerManagedResource.java +++ b/quarkus-test-containers/src/main/java/io/quarkus/test/services/containers/DockerContainerManagedResource.java @@ -24,6 +24,7 @@ import io.quarkus.test.bootstrap.ManagedResource; import io.quarkus.test.bootstrap.Protocol; import io.quarkus.test.bootstrap.ServiceContext; +import io.quarkus.test.logging.Log; import io.quarkus.test.logging.LoggingHandler; import io.quarkus.test.logging.TestContainersLoggingHandler; import io.quarkus.test.services.URILike; @@ -119,7 +120,7 @@ private void doStart() { innerContainer.start(); } catch (Exception ex) { stop(); - + loggingHandler.logs().forEach(Log::info); throw ex; } } diff --git a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/BaseKafkaContainerManagedResource.java b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/BaseKafkaContainerManagedResource.java index ff52fa450..7691d9535 100644 --- a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/BaseKafkaContainerManagedResource.java +++ b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/BaseKafkaContainerManagedResource.java @@ -7,12 +7,13 @@ import org.testcontainers.utility.MountableFile; import io.quarkus.test.bootstrap.KafkaService; +import io.quarkus.test.logging.Log; import io.quarkus.test.logging.TestContainersLoggingHandler; public abstract class BaseKafkaContainerManagedResource extends DockerContainerManagedResource { - private static final String SERVER_PROPERTIES = "server.properties"; - private static final String EXPECTED_LOG = ".*started \\(kafka.server.KafkaServer\\).*"; + private static final String SERVER_PROPERTIES = "kraft/server.properties"; + private static final String EXPECTED_LOG = ".*started .*kafka.server.Kafka.*Server.*"; protected final KafkaContainerManagedResourceBuilder model; @@ -73,6 +74,7 @@ protected GenericContainer initContainer() { String kafkaConfigPath = model.getKafkaConfigPath(); if (StringUtils.isNotEmpty(getServerProperties())) { + Log.info("Copying file %s to %s ", getServerProperties(), kafkaConfigPath + SERVER_PROPERTIES); kafkaContainer.withCopyFileToContainer(MountableFile.forClasspathResource(getServerProperties()), kafkaConfigPath + SERVER_PROPERTIES); } diff --git a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/OpenShiftStrimziKafkaContainerManagedResource.java b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/OpenShiftStrimziKafkaContainerManagedResource.java index 337c52fa6..95389fe78 100644 --- a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/OpenShiftStrimziKafkaContainerManagedResource.java +++ b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/OpenShiftStrimziKafkaContainerManagedResource.java @@ -28,7 +28,7 @@ public class OpenShiftStrimziKafkaContainerManagedResource implements ManagedRes private static final String REGISTRY_DEPLOYMENT_TEMPLATE_PROPERTY_DEFAULT = "/registry-deployment-template.yml"; private static final String REGISTRY_DEPLOYMENT = "registry.yml"; - private static final String EXPECTED_LOG = "started (kafka.server.KafkaServer)"; + private static final String EXPECTED_LOG = "Kafka Server started"; private static final int HTTP_PORT = 9092; diff --git a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/StrimziKafkaContainerManagedResource.java b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/StrimziKafkaContainerManagedResource.java index 450cb3912..5a6ce9cb3 100644 --- a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/StrimziKafkaContainerManagedResource.java +++ b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/StrimziKafkaContainerManagedResource.java @@ -42,12 +42,12 @@ public URILike getURI(Protocol protocol) { @Override protected GenericContainer initKafkaContainer() { - ExtendedStrimziKafkaContainer container = new ExtendedStrimziKafkaContainer(getKafkaImageName(), getKafkaVersion()); + ExtendedStrimziKafkaContainer container = new ExtendedStrimziKafkaContainer(getKafkaImageName(), getKafkaVersion()) + .enableKraftMode(); if (StringUtils.isNotEmpty(getServerProperties())) { container.useCustomServerProperties(); } container.withCreateContainerCmdModifier(cmd -> cmd.withName(DockerUtils.generateDockerContainerName())); - return container; } diff --git a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/strimzi/ExtendedStrimziKafkaContainer.java b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/strimzi/ExtendedStrimziKafkaContainer.java index bb717a767..b651f7045 100644 --- a/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/strimzi/ExtendedStrimziKafkaContainer.java +++ b/quarkus-test-service-kafka/src/main/java/io/quarkus/test/services/containers/strimzi/ExtendedStrimziKafkaContainer.java @@ -1,23 +1,25 @@ package io.quarkus.test.services.containers.strimzi; -import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import org.testcontainers.images.builder.Transferable; import com.github.dockerjava.api.command.InspectContainerResponse; +import io.quarkus.test.logging.Log; import io.quarkus.test.services.containers.model.KafkaVendor; import io.strimzi.test.container.StrimziKafkaContainer; /** * Extend the functionality of io.strimzi.StrimziKafkaContainer with: * - Do not overwrite parameters of server.properties. - * */ public class ExtendedStrimziKafkaContainer extends StrimziKafkaContainer { private static final String KAFKA_MAPPED_PORT = "${KAFKA_MAPPED_PORT}"; private static final int ALLOW_EXEC = 700; + private static final String TESTCONTAINERS_SCRIPT = "/testcontainers_start.sh"; private boolean useCustomServerProperties = false; @@ -31,18 +33,39 @@ public void useCustomServerProperties() { @Override protected void containerIsStarting(InspectContainerResponse containerInfo, boolean reused) { + Log.info("Starting container using custom server properties"); if (useCustomServerProperties) { + List script = new ArrayList<>(); + script.add("#!/bin/bash"); + script.add("set -euv"); int kafkaExposedPort = this.getMappedPort(KafkaVendor.STRIMZI.getPort()); - - String command = "#!/bin/bash \n"; - command = command + "sed 's/" + KAFKA_MAPPED_PORT + "/" + kafkaExposedPort + "/g' " - + "config/server.properties > /tmp/effective_server.properties &\n"; - command = command + "bin/zookeeper-server-start.sh config/zookeeper.properties &\n"; - command = command + "bin/kafka-server-start.sh /tmp/effective_server.properties"; - this.copyFileToContainer(Transferable.of(command.getBytes(StandardCharsets.UTF_8), ALLOW_EXEC), - "/testcontainers_start.sh"); + script.add("sed 's/" + KAFKA_MAPPED_PORT + "/" + kafkaExposedPort + "/g' " + + "config/kraft/server.properties > /tmp/effective_server.properties"); + script.add("KAFKA_CLUSTER_ID=\"$(bin/kafka-storage.sh random-uuid)\""); + String storageFormat = "/opt/kafka/bin/kafka-storage.sh format" + + " -t ${KAFKA_CLUSTER_ID}" + + " -c /tmp/effective_server.properties"; + script.add(storageFormat); + script.add("bin/kafka-server-start.sh /tmp/effective_server.properties"); + this.copyFileToContainer(Transferable.of(String.join("\n", script), ALLOW_EXEC), TESTCONTAINERS_SCRIPT); } else { + Log.info("Starting container using standard server properties"); + // we do not process credentials here, since SASL always used together with custom properties + // see StrimziKafkaContainerManagedResource#getServerProperties super.containerIsStarting(containerInfo, reused); + // if that is to change, we will need to copy script from test containers, modify it and copy back again } } + + /** + * The code below requires an explanation. + * StrimziKafkaContainer has a special method which makes it use kraft mode (without a zookeeper) + * Container quay.io/strimzi/kafka requires for broker.id and node.id to have the same value in kraft mode, + * and for some reason strimzi class always overwrites broker id (to 0 by default) + * since config/kraft/server.properties contains node.id=1, we have to use this value + */ + public ExtendedStrimziKafkaContainer enableKraftMode() { + return (ExtendedStrimziKafkaContainer) super.withKraft() + .withBrokerId(1); + } } diff --git a/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl-ssl.properties b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl-ssl.properties new file mode 100644 index 000000000..309db0d1e --- /dev/null +++ b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl-ssl.properties @@ -0,0 +1,152 @@ +# 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. + +# see kafka.server.KafkaConfig for additional details and defaults + +############################# Server Basics ############################# + +# The role of this server. Setting this puts us in KRaft mode +process.roles=broker,controller + +# The node id associated with this instance's roles +node.id=1 + +# The connect string for the controller quorum +controller.quorum.voters=1@localhost:9094 + +############################# Socket Server Settings ############################# + +# The address the socket server listens on. It will get the value returned from +# java.net.InetAddress.getCanonicalHostName() if not configured. +# FORMAT: +# listeners = listener_name://host_name:port +# EXAMPLE: +# listeners = PLAINTEXT://your.host.name:9092 +#listeners=PLAINTEXT://:9092 +listeners=BROKER://0.0.0.0:9093,SASL_SSL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094 + +# Name of listener used for communication between brokers. +inter.broker.listener.name=BROKER + +# Hostname and port the broker will advertise to producers and consumers. If not set, +# it uses the value for "listeners" if configured. Otherwise, it will use the value +# returned from java.net.InetAddress.getCanonicalHostName(). +#advertised.listeners=PLAINTEXT://your.host.name:9092 +advertised.listeners=SASL_SSL://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 + +# A comma-separated list of the names of the listeners used by the controller. +# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol +# This is required if running in KRaft mode. +controller.listener.names=CONTROLLER + +# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details +#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL +listener.security.protocol.map=BROKER:PLAINTEXT,SASL_SSL:SASL_SSL,CONTROLLER:PLAINTEXT + +# The number of threads that the server uses for receiving requests from the network and sending responses to the network +num.network.threads=3 + +# The number of threads that the server uses for processing requests, which may include disk I/O +num.io.threads=8 + +# The send buffer (SO_SNDBUF) used by the socket server +socket.send.buffer.bytes=102400 + +# The receive buffer (SO_RCVBUF) used by the socket server +socket.receive.buffer.bytes=102400 + +# The maximum size of a request that the socket server will accept (protection against OOM) +socket.request.max.bytes=104857600 + +############################# SASL_SSL Settings ############################# + +sasl.enabled.mechanisms=SCRAM-SHA-512 +sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 + +# Password must have at least 32 characters to work in FIPS-enabled environment +# see https://strimzi.io/blog/2023/01/25/running-apache-kafka-on-fips-enabled-kubernetes-cluster/ +listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="client" password="client-secret12345678912345678912"; + +############################# SSL ############################# + +ssl.keystore.location=/opt/kafka/config/strimzi-server-ssl-keystore.p12 +ssl.keystore.password=top-secret +ssl.keystore.type=PKCS12 +ssl.key.password=top-secret +ssl.truststore.location=/opt/kafka/config/strimzi-server-ssl-truststore.p12 +ssl.truststore.password=top-secret +ssl.truststore.type=PKCS12 +ssl.endpoint.identification.algorithm=https +ssl.client.auth=required + + +############################# Log Basics ############################# + +# A comma separated list of directories under which to store log files +log.dirs=/tmp/kraft-combined-logs + +# The default number of log partitions per topic. More partitions allow greater +# parallelism for consumption, but this will also result in more files across +# the brokers. +num.partitions=1 + +# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. +# This value is recommended to be increased for installations with data dirs located in RAID array. +num.recovery.threads.per.data.dir=1 + +############################# Internal Topic Settings ############################# +# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" +# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. +offsets.topic.replication.factor=1 +transaction.state.log.replication.factor=1 +transaction.state.log.min.isr=1 + +############################# Log Flush Policy ############################# + +# Messages are immediately written to the filesystem but by default we only fsync() to sync +# the OS cache lazily. The following configurations control the flush of data to disk. +# There are a few important trade-offs here: +# 1. Durability: Unflushed data may be lost if you are not using replication. +# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. +# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. +# The settings below allow one to configure the flush policy to flush data after a period of time or +# every N messages (or both). This can be done globally and overridden on a per-topic basis. + +# The number of messages to accept before forcing a flush of data to disk +#log.flush.interval.messages=10000 + +# The maximum amount of time a message can sit in a log before we force a flush +#log.flush.interval.ms=1000 + +############################# Log Retention Policy ############################# + +# The following configurations control the disposal of log segments. The policy can +# be set to delete segments after a period of time, or after a given size has accumulated. +# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens +# from the end of the log. + +# The minimum age of a log file to be eligible for deletion due to age +log.retention.hours=168 + +# A size-based retention policy for logs. Segments are pruned from the log unless the remaining +# segments drop below log.retention.bytes. Functions independently of log.retention.hours. +#log.retention.bytes=1073741824 + +# The maximum size of a log segment file. When this size is reached a new log segment will be created. +log.segment.bytes=1073741824 + +# The interval at which log segments are checked to see if they can be deleted according +# to the retention policies +log.retention.check.interval.ms=300000 diff --git a/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl.properties b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl.properties index 85d0e60f6..4a55cc867 100644 --- a/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl.properties +++ b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-sasl.properties @@ -13,35 +13,51 @@ # See the License for the specific language governing permissions and # limitations under the License. -# see kafka.server.KafkaConfig for additional details and defaults +# +# This configuration file is intended for use in KRaft mode, where +# Apache ZooKeeper is not present. +# ############################# Server Basics ############################# -# The id of the broker. This must be set to a unique integer for each broker. -broker.id=0 +# The role of this server. Setting this puts us in KRaft mode +process.roles=broker,controller + +# The node id associated with this instance's roles +node.id=1 + +# The connect string for the controller quorum +controller.quorum.voters=1@localhost:9094 ############################# Socket Server Settings ############################# -# The address the socket server listens on. It will get the value returned from -# java.net.InetAddress.getCanonicalHostName() if not configured. +# The address the socket server listens on. +# Combined nodes (i.e. those with `process.roles=broker,controller`) must list the controller listener here at a minimum. +# If the broker listener is not defined, the default listener will use a host name that is equal to the value of java.net.InetAddress.getCanonicalHostName(), +# with PLAINTEXT listener name, and port 9092. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 -#listeners=PLAINTEXT://:9092 -listeners=BROKER://0.0.0.0:9093,SASL_PLAINTEXT://0.0.0.0:9092 +listeners=BROKER://0.0.0.0:9093,SASL_PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094 +# Name of listener used for communication between brokers. +inter.broker.listener.name=BROKER # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 -advertised.listeners=SASL_PLAINTEXT://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 +advertised.listeners=SASL_PLAINTEXT://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 + +# A comma-separated list of the names of the listeners used by the controller. +# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol +# This is required if running in KRaft mode. +controller.listener.names=CONTROLLER # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details -#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL -listener.security.protocol.map=BROKER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT +listener.security.protocol.map=BROKER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 @@ -58,10 +74,6 @@ socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 - -inter.broker.listener.name=BROKER - - #### SASL #### sasl.enabled.mechanisms=PLAIN @@ -78,7 +90,7 @@ listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.secu ############################# Log Basics ############################# # A comma separated list of directories under which to store log files -log.dirs=/tmp/kafka-logs +log.dirs=/tmp/kraft-combined-logs # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across @@ -133,25 +145,3 @@ log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 - -############################# Zookeeper ############################# - -# Zookeeper connection string (see zookeeper docs for details). -# This is a comma separated host:port pairs, each corresponding to a zk -# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". -# You can also append an optional chroot string to the urls to specify the -# root directory for all kafka znodes. -zookeeper.connect=localhost:2181 - -# Timeout in ms for connecting to zookeeper -zookeeper.connection.timeout.ms=45000 - - -############################# Group Coordinator Settings ############################# - -# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. -# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. -# The default value for this is 3 seconds. -# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. -# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. -group.initial.rebalance.delay.ms=0 diff --git a/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-ssl.properties b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-ssl.properties index e17517338..3ae1cf76c 100644 --- a/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-ssl.properties +++ b/quarkus-test-service-kafka/src/main/resources/strimzi-default-server-ssl.properties @@ -17,8 +17,14 @@ ############################# Server Basics ############################# -# The id of the broker. This must be set to a unique integer for each broker. -broker.id=0 +# The role of this server. Setting this puts us in KRaft mode +process.roles=broker,controller + +# The node id associated with this instance's roles +node.id=1 + +# The connect string for the controller quorum +controller.quorum.voters=1@localhost:9094 ############################# Socket Server Settings ############################# @@ -29,19 +35,25 @@ broker.id=0 # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092 -listeners=BROKER://0.0.0.0:9093,SSL://0.0.0.0:9092 - +listeners=BROKER://0.0.0.0:9093,SSL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094 +# Name of listener used for communication between brokers. +inter.broker.listener.name=BROKER # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 -advertised.listeners=SSL://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 +advertised.listeners=SSL://localhost:${KAFKA_MAPPED_PORT},BROKER://localhost:9093 + +# A comma-separated list of the names of the listeners used by the controller. +# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol +# This is required if running in KRaft mode. +controller.listener.names=CONTROLLER # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL -listener.security.protocol.map=BROKER:PLAINTEXT,SSL:SSL +listener.security.protocol.map=BROKER:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 @@ -58,8 +70,6 @@ socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 -inter.broker.listener.name=BROKER - #### SSL #### ssl.keystore.location=/opt/kafka/config/strimzi-default-server-ssl-keystore.p12 @@ -75,7 +85,7 @@ ssl.endpoint.identification.algorithm= ############################# Log Basics ############################# # A comma separated list of directories under which to store log files -log.dirs=/tmp/kafka-logs +log.dirs=/tmp/kraft-combined-logs # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across @@ -130,25 +140,3 @@ log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 - -############################# Zookeeper ############################# - -# Zookeeper connection string (see zookeeper docs for details). -# This is a comma separated host:port pairs, each corresponding to a zk -# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". -# You can also append an optional chroot string to the urls to specify the -# root directory for all kafka znodes. -zookeeper.connect=localhost:2181 - -# Timeout in ms for connecting to zookeeper -zookeeper.connection.timeout.ms=45000 - - -############################# Group Coordinator Settings ############################# - -# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. -# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. -# The default value for this is 3 seconds. -# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. -# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. -group.initial.rebalance.delay.ms=0 diff --git a/quarkus-test-service-kafka/src/main/resources/strimzi-deployment-template.yml b/quarkus-test-service-kafka/src/main/resources/strimzi-deployment-template.yml index 183d6ec30..bee12bf93 100644 --- a/quarkus-test-service-kafka/src/main/resources/strimzi-deployment-template.yml +++ b/quarkus-test-service-kafka/src/main/resources/strimzi-deployment-template.yml @@ -19,17 +19,18 @@ items: status: loadBalancer: {} - apiVersion: v1 - kind: Service + kind: ConfigMap metadata: - name: zookeeper-service labels: - app: zookeeper - spec: - ports: - - port: 2181 - name: client - selector: - app: zookeeper + app: ${SERVICE_NAME} + name: "start-script" + data: + start.sh: |- + #!/bin/bash + set -euv + KAFKA_CLUSTER_ID="$(/opt/kafka/bin/kafka-storage.sh random-uuid)" + /opt/kafka/bin/kafka-storage.sh format -t ${KAFKA_CLUSTER_ID} -c config/kraft/server.properties + /opt/kafka/bin/kafka-server-start.sh config/kraft/server.properties --override listeners=PLAINTEXT://0.0.0.0:${KAFKA_PORT},CONTROLLER://localhost:9093 --override advertised.listeners=PLAINTEXT://${SERVICE_NAME}:${KAFKA_PORT} - apiVersion: "apps.openshift.io/v1" kind: "DeploymentConfig" metadata: @@ -49,35 +50,23 @@ items: - name: ${SERVICE_NAME}-container image: ${IMAGE}:${VERSION} imagePullPolicy: IfNotPresent - command: [ "/bin/sh" ] - args: [ "-c", "bin/kafka-server-start.sh config/server.properties --override listeners=PLAINTEXT://0.0.0.0:${KAFKA_PORT} --override advertised.listeners=PLAINTEXT://${SERVICE_NAME}:${KAFKA_PORT} --override zookeeper.connect=zookeeper-service:2181" ] + command: + - "/tmp/start.sh" env: - name: "LOG_DIR" value: "/tmp" ports: - containerPort: ${KAFKA_PORT} resources: {} + volumeMounts: + - name: script-volume + mountPath: "/tmp/start.sh" + subPath: "start.sh" + volumes: + - name: script-volume + configMap: + #Our template engine turns octal to decimal(for some reason) so we have to use decimal here + defaultMode: 365 # that is rxrxrx or 0555. + name: start-script triggers: - type: "ConfigChange" - - apiVersion: apps/v1 - kind: Deployment - metadata: - name: zookeeper - spec: - replicas: 1 - selector: - matchLabels: - app: zookeeper - template: - metadata: - labels: - app: zookeeper - spec: - containers: - - name: k8s-zookeeper - image: quay.io/debezium/zookeeper - ports: - - containerPort: 2181 - name: client - restartPolicy: Always - status: { } \ No newline at end of file