(kafka) branch trunk updated: MINOR: Validate the node id inside LocalTieredStorage#nodeId (#23020)

[chia7712]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 081b9d59a6f MINOR: Validate the node id inside 
LocalTieredStorage#nodeId (#23020)
081b9d59a6f is described below

commit 081b9d59a6fa4741f622902963ca5790db3546bd
Author: Ming-Yen Chung <mingyen...@gmail.com>
AuthorDate: Thu Aug 6 01:21:44 2026 +0800

    MINOR: Validate the node id inside LocalTieredStorage#nodeId (#23020)
    
    `nodeId()` returned `null` when neither `node.id` nor `broker.id` was
    set, which left the caller responsible for the check. Throwing inside
    `nodeId()` lets it return `int` instead of `Integer`, matching
    `TopicBasedRemoteLogMetadataManagerConfig#nodeId`.
    
    Reviewers: Chia-Ping Tsai <chia7...@gmail.com>, Ken Huang
     <s7133...@gmail.com>
---
 .../log/remote/storage/LocalTieredStorage.java       | 20 ++++++++------------
 1 file changed, 8 insertions(+), 12 deletions(-)

diff --git 
a/storage/src/testFixtures/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java
 
b/storage/src/testFixtures/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java
index 76e925c3639..de81fc7aa5c 100644
--- 
a/storage/src/testFixtures/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java
+++ 
b/storage/src/testFixtures/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java
@@ -227,16 +227,19 @@ public final class LocalTieredStorage implements 
RemoteStorageManager {
         this.storageListeners.add(listener);
     }
 
-    private Integer nodeId(final Map<String, ?> configs) {
+    private int nodeId(final Map<String, ?> configs) {
         final Integer nodeId = (Integer) configs.get(NODE_ID);
         if (nodeId != null) {
             return nodeId;
         }
         final Integer brokerId = (Integer) configs.get(BROKER_ID);
-        if (brokerId != null) {
-            logger.warn("The '{}' config is deprecated and will no longer be 
read in Apache Kafka 5.0. Please use '{}' instead.",
-                    BROKER_ID, NODE_ID);
+        if (brokerId == null) {
+            throw new InvalidConfigurationException(format(
+                    "Both %s and %s configs are missing. Please configure %s 
to use the LocalTieredStorage manager.",
+                    NODE_ID, BROKER_ID, NODE_ID));
         }
+        logger.warn("The '{}' config is deprecated and will no longer be read 
in Apache Kafka 5.0. Please use '{}' instead.",
+                BROKER_ID, NODE_ID);
         return brokerId;
     }
 
@@ -252,15 +255,8 @@ public final class LocalTieredStorage implements 
RemoteStorageManager {
         final String shouldDeleteOnClose = (String) 
configs.get(DELETE_ON_CLOSE_CONFIG);
         final String transfererClass = (String) 
configs.get(TRANSFERER_CLASS_CONFIG);
         final String isDeleteEnabled = (String) 
configs.get(ENABLE_DELETE_API_CONFIG);
-        final Integer nodeIdInt = nodeId(configs);
-
-        if (nodeIdInt == null) {
-            throw new InvalidConfigurationException(format(
-                    "Both %s and %s configs are missing. Please configure %s 
to use the LocalTieredStorage manager.",
-                    NODE_ID, BROKER_ID, NODE_ID));
-        }
 
-        brokerId = nodeIdInt;
+        brokerId = nodeId(configs);
         logger = new LogContext(format("[LocalTieredStorage Id=%d] ", 
brokerId)).logger(this.getClass());
 
         if (shouldDeleteOnClose != null) {