From a75e141b418355c202f8d30d16bcef8a5e701ea4 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 12 Aug 2024 14:25:26 -0400 Subject: [PATCH] GH-3416: Upgrading to kafka-client 3.8.0 Fixes: #3416 --- build.gradle | 2 +- .../kafka/test/EmbeddedKafkaKraftBroker.java | 11 ++++++----- .../kafka/test/EmbeddedKafkaZKBroker.java | 19 ++++++++++--------- .../StringOrBytesSerializerTests.java | 8 +++++--- 4 files changed, 22 insertions(+), 18 deletions(-) diff --git a/build.gradle b/build.gradle index 9a2d20b8..47ac1e45 100644 --- a/build.gradle +++ b/build.gradle @@ -58,7 +58,7 @@ ext { jaywayJsonPathVersion = '2.9.0' junit4Version = '4.13.2' junitJupiterVersion = '5.11.0-RC1' - kafkaVersion = '3.7.1' + kafkaVersion = '3.8.0' kotlinCoroutinesVersion = '1.8.1' log4jVersion = '2.23.1' micrometerDocsVersion = '1.0.3' diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java index e5635fbd..6908edc1 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java @@ -70,6 +70,7 @@ import kafka.testkit.TestKitNodes; * @author Nakul Mishra * @author Pawel Lozinski * @author Adrian Chlebosz + * @author Soby Chacko * * @since 3.1 */ @@ -252,15 +253,15 @@ public class EmbeddedKafkaKraftBroker implements EmbeddedKafkaBroker { } private void addDefaultBrokerPropsIfAbsent() { - this.brokerProperties.putIfAbsent(KafkaConfig.DeleteTopicEnableProp(), "true"); - this.brokerProperties.putIfAbsent(KafkaConfig.GroupInitialRebalanceDelayMsProp(), "0"); - this.brokerProperties.putIfAbsent(KafkaConfig.OffsetsTopicReplicationFactorProp(), "" + this.count); - this.brokerProperties.putIfAbsent(KafkaConfig.NumPartitionsProp(), "" + this.partitionsPerTopic); + this.brokerProperties.putIfAbsent("delete.topic.enable", "true"); + this.brokerProperties.putIfAbsent("group.initial.rebalance.delay.ms", "0"); + this.brokerProperties.putIfAbsent("offsets.topic.replication.factor", "" + this.count); + this.brokerProperties.putIfAbsent("num.partitions", "" + this.partitionsPerTopic); } private void logDir(Properties brokerConfigProperties) { try { - brokerConfigProperties.put(KafkaConfig.LogDirProp(), + brokerConfigProperties.put("log.dir", Files.createTempDirectory("spring.kafka." + UUID.randomUUID()).toString()); } catch (IOException e) { diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java index cd64cefb..11836591 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2023 the original author or authors. + * Copyright 2018-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -87,6 +87,7 @@ import kafka.zookeeper.ZooKeeperClient; * @author Nakul Mishra * @author Pawel Lozinski * @author Adrian Chlebosz + * @author Soby Chacko * * @since 2.2 */ @@ -302,17 +303,17 @@ public class EmbeddedKafkaZKBroker implements EmbeddedKafkaBroker { } this.zkConnect = LOOPBACK + ":" + this.zookeeper.getPort(); this.kafkaServers.clear(); - boolean userLogDir = this.brokerProperties.get(KafkaConfig.LogDirProp()) != null && this.count == 1; + boolean userLogDir = this.brokerProperties.get("log.dir") != null && this.count == 1; for (int i = 0; i < this.count; i++) { Properties brokerConfigProperties = createBrokerProperties(i); - brokerConfigProperties.setProperty(KafkaConfig.ReplicaSocketTimeoutMsProp(), "1000"); - brokerConfigProperties.setProperty(KafkaConfig.ControllerSocketTimeoutMsProp(), "1000"); - brokerConfigProperties.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp(), "1"); - brokerConfigProperties.setProperty(KafkaConfig.ReplicaHighWatermarkCheckpointIntervalMsProp(), + brokerConfigProperties.setProperty("replica.socket.timeout.ms", "1000"); + brokerConfigProperties.setProperty("controller.socket.timeout.ms", "1000"); + brokerConfigProperties.setProperty("offsets.topic.replication.factor", "1"); + brokerConfigProperties.setProperty("replica.high.watermark.checkpoint.interval.ms", String.valueOf(Long.MAX_VALUE)); this.brokerProperties.forEach(brokerConfigProperties::put); - if (!this.brokerProperties.containsKey(KafkaConfig.NumPartitionsProp())) { - brokerConfigProperties.setProperty(KafkaConfig.NumPartitionsProp(), "" + this.partitionsPerTopic); + if (!this.brokerProperties.containsKey("num.partitions")) { + brokerConfigProperties.setProperty("num.partitions", "" + this.partitionsPerTopic); } if (!userLogDir) { logDir(brokerConfigProperties); @@ -337,7 +338,7 @@ public class EmbeddedKafkaZKBroker implements EmbeddedKafkaBroker { private void logDir(Properties brokerConfigProperties) { try { - brokerConfigProperties.put(KafkaConfig.LogDirProp(), + brokerConfigProperties.put("log.dir", Files.createTempDirectory("spring.kafka." + UUID.randomUUID()).toString()); } catch (IOException e) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/StringOrBytesSerializerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/StringOrBytesSerializerTests.java index 4ba81b73..ab123bed 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/StringOrBytesSerializerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/StringOrBytesSerializerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,7 @@ package org.springframework.kafka.support.serializer; import static org.assertj.core.api.Assertions.assertThat; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.Map; @@ -28,6 +29,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; /** * @author Gary Russell + * @author Soby Chacko * @since 2.3 * */ @@ -45,10 +47,10 @@ public class StringOrBytesSerializerTests { Bytes bytes = Bytes.wrap("baz".getBytes()); out = serializer.serialize("x", bytes); assertThat(out).isEqualTo("baz".getBytes()); - assertThat(KafkaTestUtils.getPropertyValue(serializer, "stringSerializer.encoding")).isEqualTo("UTF-8"); + assertThat(KafkaTestUtils.getPropertyValue(serializer, "stringSerializer.encoding")).isEqualTo(StandardCharsets.UTF_8); Map configs = Collections.singletonMap("serializer.encoding", "UTF-16"); serializer.configure(configs, false); - assertThat(KafkaTestUtils.getPropertyValue(serializer, "stringSerializer.encoding")).isEqualTo("UTF-16"); + assertThat(KafkaTestUtils.getPropertyValue(serializer, "stringSerializer.encoding")).isEqualTo(StandardCharsets.UTF_16); } }