Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3416: Upgrading to kafka-client 3.8.0

Browse Source 
Fixes: #3416
This commit is contained in:
Soby Chacko
2024-08-12 14:25:26 -04:00
parent 5120260ce8
commit a75e141b41
4 changed files with 22 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

11
spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java
View File

@@ -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) {

19
spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java
View File

@@ -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) {

8
spring-kafka/src/test/java/org/springframework/kafka/support/serializer/StringOrBytesSerializerTests.java
View File

@@ -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<String, Object> 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);
}
}