Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Bump kafka.version from 3.7.1 to 3.9.0 #2814

Merged
merged 3 commits into from
Nov 20, 2024
Merged

Bump kafka.version from 3.7.1 to 3.9.0 #2814

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dba84b1
Bump kafka.version from 3.7.1 to 3.9.0
dependabot[bot] Nov 7, 2024
953dac7
Config options moved with 3.8.1
ozangunalp Oct 31, 2024
40d2510
Update EmbeddedKafkaBroker to Kafka 3.9 changes
ozangunalp Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@

<jboss-log-manager.version>3.0.6.Final</jboss-log-manager.version>

<kafka.version>3.7.1</kafka.version>
<kafka.version>3.9.0</kafka.version>

<opentelemetry.instrumentation.version>2.5.0-alpha</opentelemetry.instrumentation.version>
<opentelemetry-semconv.version>1.25.0-alpha</opentelemetry-semconv.version>
Expand Down
24 changes: 24 additions & 0 deletions smallrye-reactive-messaging-kafka-test-companion/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,30 @@
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-server</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-raft</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-group-coordinator</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-storage</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<!-- Optional dependencies for usage in Tests -->
<dependency>
<groupId>org.junit.jupiter</groupId>
Expand Down
119 changes: 77 additions & 42 deletions ...rc/main/java/io/smallrye/reactive/messaging/kafka/companion/test/EmbeddedKafkaBroker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import static io.smallrye.reactive.messaging.kafka.companion.test.EmbeddedKafkaBroker.LoggingOutputStream.loggerPrintStream;
import static org.apache.kafka.common.security.auth.SecurityProtocol.PLAINTEXT;
import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_BOOTSTRAP_VERSION;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
Expand All @@ -17,6 +16,8 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Function;
Expand All @@ -27,16 +28,20 @@
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.metadata.properties.MetaProperties;
import org.apache.kafka.metadata.properties.MetaPropertiesVersion;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.metadata.storage.Formatter;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.raft.QuorumConfig;
import org.apache.kafka.server.config.KRaftConfigs;
import org.apache.kafka.server.config.ReplicationConfigs;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.jboss.logging.Logger;

import kafka.cluster.EndPoint;
import kafka.server.KafkaConfig;
import kafka.server.KafkaRaftServer;
import kafka.tools.StorageTool;
import scala.collection.immutable.Seq;
import scala.jdk.CollectionConverters;
import scala.jdk.javaapi.StreamConverters;

/**
Expand Down Expand Up @@ -194,7 +199,7 @@ public synchronized EmbeddedKafkaBroker start() {
brokerConfigModifier.accept(properties);
}

if (properties.get(KafkaConfig.LogDirProp()) == null) {
if (properties.get(ServerLogConfigs.LOG_DIR_CONFIG) == null) {
createAndSetlogDir(properties);
}

Expand Down Expand Up @@ -296,12 +301,12 @@ public static Endpoint parseEndpoint(String listenerStr) {
public static Properties createDefaultBrokerConfig(int nodeId, Endpoint controller, Endpoint internalEndpoint,
List<Endpoint> advertisedListeners) {
Properties props = new Properties();
props.put(KafkaConfig.BrokerIdProp(), Integer.toString(nodeId));
props.put(ServerConfigs.BROKER_ID_CONFIG, Integer.toString(nodeId));

// Configure kraft
props.put(KafkaConfig.ProcessRolesProp(), "broker,controller");
props.put(KafkaConfig.ControllerListenerNamesProp(), listenerName(controller));
props.put(KafkaConfig.QuorumVotersProp(), nodeId + "@" + controller.host() + ":" + controller.port());
props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker,controller");
props.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, listenerName(controller));
props.put(QuorumConfig.QUORUM_VOTERS_CONFIG, nodeId + "@" + controller.host() + ":" + controller.port());

// Configure listeners
Map<String, Endpoint> listeners = advertisedListeners.stream()
Expand All @@ -314,7 +319,7 @@ public static Properties createDefaultBrokerConfig(int nodeId, Endpoint controll
.map(EmbeddedKafkaBroker::toListenerString)
.distinct()
.collect(Collectors.joining(","));
props.put(KafkaConfig.ListenersProp(), listenersString);
props.put(SocketServerConfigs.LISTENERS_CONFIG, listenersString);

// Find a PLAINTEXT listener
Endpoint plaintextEndpoint = advertisedListeners.stream()
Expand All @@ -327,53 +332,83 @@ public static Properties createDefaultBrokerConfig(int nodeId, Endpoint controll
.distinct()
.collect(Collectors.joining(","));
if (!Utils.isBlank(advertisedListenersString)) {
props.put(KafkaConfig.AdvertisedListenersProp(), advertisedListenersString);
props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, advertisedListenersString);
}

// Configure security protocol map
String securityProtocolMap = listeners.values().stream()
.map(EmbeddedKafkaBroker::toProtocolMap)
.distinct()
.collect(Collectors.joining(","));
props.put(KafkaConfig.ListenerSecurityProtocolMapProp(), securityProtocolMap);
props.put(KafkaConfig.InterBrokerListenerNameProp(), listenerName(plaintextEndpoint));
props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, securityProtocolMap);
props.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, listenerName(plaintextEndpoint));

// Configure static default props
props.put(KafkaConfig.ReplicaSocketTimeoutMsProp(), "1000");
props.put(KafkaConfig.ReplicaHighWatermarkCheckpointIntervalMsProp(), String.valueOf(Long.MAX_VALUE));
props.put(KafkaConfig.ControllerSocketTimeoutMsProp(), "1000");
props.put(KafkaConfig.ControlledShutdownEnableProp(), Boolean.toString(false));
props.put(KafkaConfig.ControlledShutdownRetryBackoffMsProp(), "100");
props.put(KafkaConfig.DeleteTopicEnableProp(), Boolean.toString(true));
props.put(KafkaConfig.LogDeleteDelayMsProp(), "1000");
props.put(KafkaConfig.LogCleanerDedupeBufferSizeProp(), "2097152");
props.put(KafkaConfig.LogMessageTimestampDifferenceMaxMsProp(), String.valueOf(Long.MAX_VALUE));
props.put(KafkaConfig.OffsetsTopicReplicationFactorProp(), "1");
props.put(KafkaConfig.OffsetsTopicPartitionsProp(), "5");
props.put(KafkaConfig.GroupInitialRebalanceDelayMsProp(), "0");
props.put(KafkaConfig.NumPartitionsProp(), "1");
props.put(KafkaConfig.DefaultReplicationFactorProp(), "1");
props.put(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG, "1000");
props.put(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG, String.valueOf(Long.MAX_VALUE));
props.put(ReplicationConfigs.CONTROLLER_SOCKET_TIMEOUT_MS_CONFIG, "1000");
props.put(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, Boolean.toString(false));
props.put(ServerConfigs.CONTROLLED_SHUTDOWN_RETRY_BACKOFF_MS_CONFIG, "100");
props.put(ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG, Boolean.toString(true));
props.put(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, "1000");
props.put(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, "2097152");
props.put(ServerLogConfigs.LOG_MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_DOC, String.valueOf(Long.MAX_VALUE));
props.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, "1");
props.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, "5");
props.put(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, "0");
props.put(ServerLogConfigs.NUM_PARTITIONS_CONFIG, "1");
props.put(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, "1");

return props;
}

public static KafkaConfig formatStorageFromConfig(Properties properties, String clusterId, boolean ignoreFormatted) {
KafkaConfig config = KafkaConfig.fromProps(properties, false);
Seq<String> directories = StorageTool.configToLogDirectories(config);
MetaProperties metaProperties = StorageTool.buildMetadataProperties(clusterId, config);
StorageTool.formatCommand(loggerPrintStream(LOGGER), directories, metaProperties, MINIMUM_BOOTSTRAP_VERSION,
ignoreFormatted);
Formatter formatter = new Formatter();
formatter.setClusterId(clusterId)
.setNodeId(config.nodeId())
.setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled())
.setIgnoreFormatted(ignoreFormatted)
.setControllerListenerName(config.controllerListenerNames().head())
.setMetadataLogDirectory(config.metadataLogDir());
configToLogDirectories(config).forEach(formatter::addDirectory);
try {
formatter.run();
} catch (Exception e) {
throw new RuntimeException(e);
}
return config;
}

public static void formatStorage(List<String> directories, String clusterId, int nodeId, boolean ignoreFormatted) {
MetaProperties metaProperties = new MetaProperties.Builder()
.setVersion(MetaPropertiesVersion.V1)
.setClusterId(clusterId)
static Set<String> configToLogDirectories(KafkaConfig config) {
TreeSet<String> dirs = new TreeSet<>();
config.logDirs().foreach(dirs::add);
String metadataLogDir = config.metadataLogDir();
if (metadataLogDir != null) {
dirs.add(metadataLogDir);
}
return dirs;
}

public static void formatStorage(List<String> directories, String controllerListenerName,
String metadataLogDirectory,
String clusterId, int nodeId, boolean ignoreFormatted) {
Formatter formatter = new Formatter();
formatter.setClusterId(clusterId)
.setNodeId(nodeId)
.build();
Seq<String> dirs = CollectionConverters.ListHasAsScala(directories).asScala().toSeq();
StorageTool.formatCommand(loggerPrintStream(LOGGER), dirs, metaProperties, MINIMUM_BOOTSTRAP_VERSION, ignoreFormatted);
.setIgnoreFormatted(ignoreFormatted)
.setControllerListenerName(controllerListenerName)
.setMetadataLogDirectory(metadataLogDirectory);
directories.forEach(formatter::addDirectory);
try {
formatter.run();
} catch (Exception e) {
throw new RuntimeException(e);
}
}

public static void formatStorage(List<String> directories, String clusterId, int nodeId, boolean ignoreFormatted) {
formatStorage(directories, "CONTROLLER", directories.get(0), clusterId, nodeId, ignoreFormatted);
}

public static KafkaRaftServer createServer(final KafkaConfig config) {
Expand All @@ -383,7 +418,7 @@ public static KafkaRaftServer createServer(final KafkaConfig config) {
}

private static String getAdvertisedListeners(KafkaConfig config) {
return StreamConverters.asJavaParStream(config.effectiveAdvertisedListeners())
return StreamConverters.asJavaParStream(config.effectiveAdvertisedBrokerListeners())
.map(EndPoint::connectionString)
.collect(Collectors.joining(","));
}
Expand All @@ -406,7 +441,7 @@ private static int getUnusedPort(int port) {

private static void createAndSetlogDir(Properties properties) {
try {
properties.put(KafkaConfig.LogDirProp(),
properties.put(ServerLogConfigs.LOG_DIR_CONFIG,
Files.createTempDirectory(COMPANION_BROKER_PREFIX + "-" + UUID.randomUUID()).toString());
} catch (IOException e) {
throw new UncheckedIOException(e);
Expand Down
4 changes: 2 additions & 2 deletions .../src/test/java/io/smallrye/reactive/messaging/kafka/companion/test/EmbeddedKafkaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,13 @@
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Endpoint;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import io.smallrye.reactive.messaging.kafka.companion.KafkaCompanion;
import io.smallrye.reactive.messaging.kafka.companion.TestTags;
import kafka.server.KafkaConfig;

@Tag(TestTags.FLAKY)
public class EmbeddedKafkaTest {
Expand All @@ -42,7 +42,7 @@ void test() {
void testWithExistingLogDir(@TempDir File dir) {
EmbeddedKafkaBroker broker = new EmbeddedKafkaBroker()
.withNodeId(0)
.withAdditionalProperties(props -> props.put(KafkaConfig.LogDirProp(), dir.toPath().toString()))
.withAdditionalProperties(props -> props.put(ServerLogConfigs.LOG_DIR_CONFIG, dir.toPath().toString()))
.withDeleteLogDirsOnClose(false);

// format storage before starting
Expand Down