This is an automated email from the ASF dual-hosted git repository.
rabreu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 2958d3ff8 Bump storm.kafka.client.version from 3.9.0 to 4.1.0 (#8243)
2958d3ff8 is described below
commit 2958d3ff8b1b3071418a339ea2584faf270bf5ac
Author: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
AuthorDate: Sun Sep 14 21:40:17 2025 +0100
Bump storm.kafka.client.version from 3.9.0 to 4.1.0 (#8243)
* Bump storm.kafka.client.version from 3.9.0 to 4.1.0
Bumps `storm.kafka.client.version` from 3.9.0 to 4.1.0.
Updates `org.apache.kafka:kafka-clients` from 3.9.0 to 4.1.0
Updates `org.apache.kafka:kafka_2.13` from 3.9.0 to 4.1.0
---
updated-dependencies:
- dependency-name: org.apache.kafka:kafka-clients
dependency-version: 4.1.0
dependency-type: direct:production
update-type: version-update:semver-major
- dependency-name: org.apache.kafka:kafka_2.13
dependency-version: 4.1.0
dependency-type: direct:production
update-type: version-update:semver-major
...
Signed-off-by: dependabot[bot] <[email protected]>
* Fixing licenses
* Adapting code to Kafka 4 API
* Adapting code to Kafka 4 API
* Adapting code to Kafka 4 API
* Adapting code to Kafka 4 API
* Fixing licenses
* Fixing licenses
* Fixing licenses
---------
Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot]
<49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Rui Abreu <[email protected]>
---
DEPENDENCY-LICENSES | 4 +-
LICENSE-binary | 4 +-
.../org/apache/storm/kafka/spout/KafkaSpout.java | 4 +-
.../spout/trident/KafkaTridentSpoutEmitter.java | 2 +-
.../java/org/apache/storm/kafka/KafkaUnit.java | 46 +++++++++++++++-------
.../storm/kafka/spout/KafkaSpoutRebalanceTest.java | 14 ++++---
.../storm/kafka/monitor/KafkaOffsetLagUtil.java | 6 ++-
pom.xml | 2 +-
8 files changed, 51 insertions(+), 31 deletions(-)
diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES
index c29d0dc6a..68976e467 100644
--- a/DEPENDENCY-LICENSES
+++ b/DEPENDENCY-LICENSES
@@ -63,7 +63,7 @@ List of third-party dependencies grouped by their license
type.
* Apache HttpComponents Core HTTP/1.1
(org.apache.httpcomponents.core5:httpcore5:5.2 -
https://hc.apache.org/httpcomponents-core-5.2.x/5.2/httpcore5/)
* Apache HttpComponents Core HTTP/2
(org.apache.httpcomponents.core5:httpcore5-h2:5.2 -
https://hc.apache.org/httpcomponents-core-5.2.x/5.2/httpcore5-h2/)
* Apache HttpCore (org.apache.httpcomponents:httpcore:4.4.16 -
http://hc.apache.org/httpcomponents-core-ga)
- * Apache Kafka (org.apache.kafka:kafka-clients:3.9.0 -
https://kafka.apache.org)
+ * Apache Kafka (org.apache.kafka:kafka-clients:4.1.0 -
https://kafka.apache.org)
* Apache Log4j API (org.apache.logging.log4j:log4j-api:2.25.1 -
https://logging.apache.org/log4j/2.x/)
* Apache Log4j Core (org.apache.logging.log4j:log4j-core:2.25.1 -
https://logging.apache.org/log4j/2.x/)
* Apache Maven Shade Plugin
(org.apache.maven.plugins:maven-shade-plugin:3.6.0 -
https://maven.apache.org/plugins/maven-shade-plugin/)
@@ -325,7 +325,7 @@ List of third-party dependencies grouped by their license
type.
BSD 2-Clause License
- * zstd-jni (com.github.luben:zstd-jni:1.5.6-4 -
https://github.com/luben/zstd-jni)
+ * zstd-jni (com.github.luben:zstd-jni:1.5.6-10 -
https://github.com/luben/zstd-jni)
BSD 3-Clause License
diff --git a/LICENSE-binary b/LICENSE-binary
index b2857692a..6a6badf45 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -700,7 +700,7 @@ The license texts of these dependencies can be found in the
licenses directory.
* Apache HBase Unsafe Wrapper
(org.apache.hbase.thirdparty:hbase-unsafe:4.1.11 -
https://hbase.apache.org/hbase-unsafe)
* Apache HttpClient (org.apache.httpcomponents:httpclient:4.5.14 -
http://hc.apache.org/httpcomponents-client-ga)
* Apache HttpCore (org.apache.httpcomponents:httpcore:4.4.16 -
http://hc.apache.org/httpcomponents-core-ga)
- * Apache Kafka (org.apache.kafka:kafka-clients:3.9.0 -
https://kafka.apache.org)
+ * Apache Kafka (org.apache.kafka:kafka-clients:4.1.0 -
https://kafka.apache.org)
* Apache Log4j API (org.apache.logging.log4j:log4j-api:2.25.1 -
https://logging.apache.org/log4j/2.x/log4j-api/)
* Apache Log4j Core (org.apache.logging.log4j:log4j-core:2.25.1 -
https://logging.apache.org/log4j/2.x/log4j-core/)
* Apache Log4j SLF4J Binding
(org.apache.logging.log4j:log4j-slf4j2-impl:2.25.1 -
https://logging.apache.org/log4j/2.x/log4j-slf4j2-impl/)
@@ -906,7 +906,7 @@ The license texts of these dependencies can be found in the
licenses directory.
BSD 2-Clause license
* dnsjava (dnsjava:dnsjava:3.6.1 - http://www.dnsjava.org)
- * zstd-jni (com.github.luben:zstd-jni:1.5.6-4 -
https://github.com/luben/zstd-jni)
+ * zstd-jni (com.github.luben:zstd-jni:1.5.6-10 -
https://github.com/luben/zstd-jni)
BSD-3-Clause
diff --git
a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
index 0fcb22a04..d66d936a2 100644
---
a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
+++
b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
@@ -219,8 +219,8 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
// leave the acked offsets and consumer position as they were to
resume where it left off
newPartitions.removeAll(previousAssignment);
for (TopicPartition newTp : newPartitions) {
- final OffsetAndMetadata committedOffset =
consumer.committed(newTp);
- final long fetchOffset = doSeek(newTp, committedOffset);
+ final Map<TopicPartition, OffsetAndMetadata> committedOffset =
consumer.committed(Collections.singleton(newTp));
+ final long fetchOffset = doSeek(newTp,
committedOffset.get(newTp));
LOG.debug("Set consumer position to [{}] for topic-partition
[{}] with [{}] and committed offset [{}]",
fetchOffset, newTp, firstPollOffsetStrategy,
committedOffset);
if (isAtLeastOnceProcessing() &&
!offsetManagers.containsKey(newTp)) {
diff --git
a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
index c9ddf392e..1c0a6ba41 100644
---
a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
+++
b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
@@ -138,7 +138,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements
Serializable {
LOG.debug("Seeking to offset [{}] for topic partition [{}]",
seekOffset, currBatchTp);
consumer.seek(currBatchTp, seekOffset);
- final ConsumerRecords<K, V> records = consumer.poll(pollTimeoutMs);
+ final ConsumerRecords<K, V> records =
consumer.poll(Duration.ofMillis(pollTimeoutMs));
LOG.debug("Polled [{}] records from Kafka.", records.count());
for (ConsumerRecord<K, V> record : records) {
diff --git
a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/KafkaUnit.java
b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/KafkaUnit.java
index 50dbdf84a..de14a300d 100644
---
a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/KafkaUnit.java
+++
b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/KafkaUnit.java
@@ -22,53 +22,69 @@ import static
org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CL
import static
org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import kafka.utils.TestUtils;
+import kafka.server.KafkaRaftServer;
+import kafka.server.Server;
+import org.apache.commons.lang.RandomStringUtils;
import org.apache.curator.test.TestingServer;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
import org.apache.storm.testing.TmpPath;
public class KafkaUnit {
private TestingServer zookeeper;
- private KafkaServer kafkaServer;
+ private Server kafkaServer;
private KafkaProducer<String, String> producer;
private AdminClient kafkaAdminClient;
- private TmpPath kafkaDir;
+ private TmpPath kafkaDir,metadata;
private static final String KAFKA_HOST = "127.0.0.1";
- private static final int KAFKA_PORT = 9092;
+ private static final int KAFKA_BROKER_PORT = 9092;
+ private static final int KAFKA_CONTROLLER_PORT = 9093;
public KafkaUnit() {
}
public void setUp() throws Exception {
- // setup ZK
- zookeeper = new TestingServer(true);
// setup Broker
kafkaDir = new
TmpPath(Files.createTempDirectory("kafka-").toAbsolutePath().toString());
+ metadata = new TmpPath(Paths.get(kafkaDir.getPath() + "/" +
"meta.properties").toString());
+ Files.createFile(metadata.getFile().toPath());
+ String content ="node.id=0" + System.lineSeparator() + "version=1" +
System.lineSeparator() + "cluster.id="+
RandomStringUtils.randomAlphanumeric(10);
+ Files.writeString(metadata.getFile().toPath(), content,
StandardOpenOption.APPEND);
+
Properties brokerProps = new Properties();
- brokerProps.setProperty("zookeeper.connect",
zookeeper.getConnectString());
brokerProps.setProperty("broker.id", "0");
brokerProps.setProperty("log.dirs", kafkaDir.getPath());
- brokerProps.setProperty("listeners",
String.format("PLAINTEXT://%s:%d", KAFKA_HOST, KAFKA_PORT));
+ String listeners = String.format("BROKER://%s:%d", KAFKA_HOST,
KAFKA_BROKER_PORT) + "," + String.format("CONTROLLER://%s:%d", KAFKA_HOST,
KAFKA_CONTROLLER_PORT);
+ brokerProps.setProperty("advertised.listeners", listeners);
+ brokerProps.setProperty("listeners", listeners);
brokerProps.setProperty("offsets.topic.replication.factor", "1");
+ brokerProps.setProperty("process.roles","broker,controller");
+
brokerProps.setProperty("controller.quorum.bootstrap.servers",String.format("%s:%d",
KAFKA_HOST, KAFKA_CONTROLLER_PORT));
+ brokerProps.setProperty("controller.listener.names","CONTROLLER");
+
brokerProps.setProperty("listener.security.protocol.map","CONTROLLER:PLAINTEXT,BROKER:PLAINTEXT");
+ brokerProps.setProperty("inter.broker.listener.name","BROKER");
+
brokerProps.setProperty("controller.quorum.voters","0@"+String.format("%s:%d",
KAFKA_HOST, KAFKA_CONTROLLER_PORT));
+
KafkaConfig config = new KafkaConfig(brokerProps);
- kafkaServer = TestUtils.createServer(config, new MockTime());
+ kafkaServer= new KafkaRaftServer(config, Time.SYSTEM);
+ kafkaServer.startup();
// setup default Producer
createProducer();
- kafkaAdminClient =
AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
KAFKA_HOST + ":" + KAFKA_PORT));
+ kafkaAdminClient =
AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
KAFKA_HOST + ":" + KAFKA_BROKER_PORT));
}
public void tearDown() throws Exception {
@@ -76,7 +92,7 @@ public class KafkaUnit {
closeProducer();
kafkaServer.shutdown();
kafkaDir.close();
- zookeeper.close();
+ metadata.close();
}
public void createTopic(String topicName) throws Exception {
@@ -86,12 +102,12 @@ public class KafkaUnit {
}
public int getKafkaPort() {
- return KAFKA_PORT;
+ return KAFKA_BROKER_PORT;
}
private void createProducer() {
Properties producerProps = new Properties();
- producerProps.setProperty(BOOTSTRAP_SERVERS_CONFIG, KAFKA_HOST + ":" +
KAFKA_PORT);
+ producerProps.setProperty(BOOTSTRAP_SERVERS_CONFIG, KAFKA_HOST + ":" +
KAFKA_BROKER_PORT);
producerProps.setProperty(KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
producerProps.setProperty(VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<>(producerProps);
diff --git
a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
index 74b0f84fc..c8847750c 100644
---
a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
+++
b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
@@ -114,9 +114,9 @@ public class KafkaSpoutRebalanceTest {
//Make the consumer return a single message for each partition
when(consumerMock.poll(any(Duration.class)))
- .thenReturn(new
ConsumerRecords<>(Collections.singletonMap(partitionThatWillBeRevoked,
SpoutWithMockedConsumerSetupHelper.createRecords(partitionThatWillBeRevoked, 0,
1))))
- .thenReturn(new
ConsumerRecords<>(Collections.singletonMap(assignedPartition,
SpoutWithMockedConsumerSetupHelper.createRecords(assignedPartition, 0, 1))))
- .thenReturn(new ConsumerRecords<>(Collections.emptyMap()));
+ .thenReturn(new
ConsumerRecords<>(Collections.singletonMap(partitionThatWillBeRevoked,
SpoutWithMockedConsumerSetupHelper.createRecords(partitionThatWillBeRevoked, 0,
1)), Map.of()))
+ .thenReturn(new
ConsumerRecords<>(Collections.singletonMap(assignedPartition,
SpoutWithMockedConsumerSetupHelper.createRecords(assignedPartition, 0, 1)),
Map.of()))
+ .thenReturn(new ConsumerRecords<>(Collections.emptyMap(),
Map.of()));
//Emit the messages
spout.nextTuple();
@@ -236,9 +236,11 @@ public class KafkaSpoutRebalanceTest {
//Set up committed so it looks like some messages have been committed
on each partition
long committedOffset = 500;
- final Answer<Object> objectAnswer = invocation -> new
OffsetAndMetadata(committedOffset);
-
lenient().doAnswer(objectAnswer).when(consumerMock).committed(assignedPartition);
- doAnswer(objectAnswer).when(consumerMock).committed(newPartition);
+ final Map<TopicPartition, OffsetAndMetadata> mapAnswer = new
HashMap<>();
+ mapAnswer.put(newPartition,new OffsetAndMetadata(committedOffset));
+ final Answer<Object> objectAnswer = invocation -> mapAnswer;
+
lenient().doAnswer(objectAnswer).when(consumerMock).committed(Collections.singleton(newPartition));
+
doAnswer(objectAnswer).when(consumerMock).committed(Collections.singleton(newPartition));
//Now rebalance and add a new partition
consumerRebalanceListener.onPartitionsRevoked(assignedPartitions);
diff --git
a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
index 77d9f78e6..fa06ffa3e 100644
---
a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
+++
b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
@@ -20,10 +20,12 @@ package org.apache.storm.kafka.monitor;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+
import net.minidev.json.JSONValue;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
@@ -171,8 +173,8 @@ public class KafkaOffsetLagUtil {
}
consumer.assign(topicPartitionList);
for (TopicPartition topicPartition : topicPartitionList) {
- OffsetAndMetadata offsetAndMetadata =
consumer.committed(topicPartition);
- long committedOffset = offsetAndMetadata != null ?
offsetAndMetadata.offset() : -1;
+ Map<TopicPartition, OffsetAndMetadata> offsetAndMetadata =
consumer.committed(Collections.singleton(topicPartition));
+ long committedOffset = offsetAndMetadata != null ?
offsetAndMetadata.get(topicPartition).offset() : -1;
consumer.seekToEnd(toArrayList(topicPartition));
result.add(new KafkaOffsetLagResult(topicPartition.topic(),
topicPartition.partition(), committedOffset,
consumer.position(topicPartition)));
diff --git a/pom.xml b/pom.xml
index df24ce00b..c3643492e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -134,7 +134,7 @@
<jackson.version>2.20.0</jackson.version>
<jackson.databind.version>2.20.0</jackson.databind.version>
- <storm.kafka.client.version>3.9.0</storm.kafka.client.version>
+ <storm.kafka.client.version>4.1.0</storm.kafka.client.version>
<testcontainers.version>1.21.3</testcontainers.version>
<!-- Java and clojure build lifecycle test properties are defined here
to avoid having to create a default profile -->
