522 - Kafka steram client upgrade initial change by sanath-madhav · Pull Request #77 · eclipse-ecsp/streambase

sanath-madhav · 2025-10-28T14:04:33Z

Please refer to our contributing docs for any questions on submitting a pull request.
Issues are required for both bug fixes and features.

Resolves #522

Describe behaviour before the change

Behaviour unchanged, updated new upgraded api changed to existing code base

Describe behaviour after the change

Behaviour unchanged, updated new upgraded api changed to existing code base
Documentation- https://github.com/HARMAN-Auto/sp-platform-productenablers-streambase/blob/main/docs/designs/kafka-client-upgrade-3.9.1.md,

Pull request checklist

I have read the CONTRIBUTING.md
My code follows the code style of this project
Tests for the changes have been added (for bug fixes / features)
All new and existing tests passed.
Docs have been reviewed and added / updated if needed (for bug fixes / features)

Does this introduce a breaking change?

Yes
No

kaushalaroraharman · 2025-10-30T08:59:39Z

@sanath-madhav Please update the required details in the PR, such as link to ticket it resolves, section for "Describe behaviour before the change", checklist, etc.

kaushalaroraharman

Comments added in-line

kaushalaroraharman · 2025-10-30T09:02:42Z

    @Override
-    public void write(final WriteBatch batch) throws RocksDBException {
-        db.write(writeOptions, batch);
+    public void write(final WriteBatchInterface batch) throws RocksDBException {


Please provide reason for this change from WriteBatch to WriteBatchInterface as the method parameter.

Also please document this on the design page here : https://github.com/HARMAN-Auto/sp-platform-productenablers-streambase/blob/main/docs/designs/kafka-client-upgrade-3.9.1.md, along with all the other required details for this upgrade.

Interface signature change

from
void write(final WriteBatch batch) throws RocksDBException
to
void write(final WriteBatchInterface batch) throws RocksDBException

kaushalaroraharman · 2025-10-30T09:08:11Z

+        if (batch instanceof WriteBatch writeBatch) {
+            db.write(writeOptions, writeBatch);
+        } else {
+            throw new IllegalArgumentException("Batch must be an instance of WriteBatch");


If any other implementation is not supported by this method, add an error level logger as well before throwing the exception and also add the type of instance received by the method.

Added error log before throwing the exception

kaushalaroraharman · 2025-10-30T09:10:17Z

+            }
        } else {
-            batch.put(key, value);
+            throw new IllegalArgumentException("Batch must be an instance of WriteBatch");


Same as above comment, add a logger and the type of instance received in this case.

Added error log before throwing the exception

kaushalaroraharman · 2025-10-30T09:19:23Z

-        effectiveConfigProps.put(KafkaConfig$.MODULE$.AutoCreateTopicsEnableProp(), true);
-        effectiveConfigProps.put(KafkaConfig$.MODULE$.MessageMaxBytesProp(), Constants.INT_1000000);
-        effectiveConfigProps.put(KafkaConfig$.MODULE$.ControlledShutdownEnableProp(), true);
+        effectiveConfigProps.put("broker.id", 0);


Per the documentation, this has been removed in favour of KRaft adoption.
There should be properties available in https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaConfig.scala, which can be used. Avoid hardcoding property names here, and use the one provided by Kafka client library. Only if not able to use from KafkaConfig class, you can create these in propertyNames file, whilst some would already be existing.

Moved to Constant file

kaushalaroraharman · 2025-10-30T09:21:28Z

        final Properties effectiveBrokerConfig = effectiveBrokerConfigFrom(brokerConfig, zookeeper);
        LOG.debug("Starting a Kafka instance on port {} ...",
-                effectiveBrokerConfig.getProperty(KafkaConfig.ListenersProp()));
+                effectiveBrokerConfig.getProperty("listeners"));


Do not hardcode property names directly in the code. Either fetch from Kafka client library, if the source has changed post upgrade, or use from PropertyNames class in stream-base.

Moved to Constant file

kaushalaroraharman · 2025-10-30T09:22:24Z

    public String brokerList() {
-        final EndPoint endPoint = ((ArraySeq<EndPoint>) kafka.advertisedListeners()).head();
+        // Kafka 3.9.1 returns ArrayBuffer instead of ArraySeq
+        final scala.collection.Seq<EndPoint> listeners = kafka.advertisedListeners();


Is there any particular reason for using the fully qualified name here instead of importing the required class?

Api has been changed

final EndPoint endPoint = ((ArrayBuffer) kafka.advertisedListeners()).head();

kaushalaroraharman · 2025-10-30T09:22:56Z

-        final EndPoint endPoint = ((ArraySeq<EndPoint>) kafka.advertisedListeners()).head();
+        // Kafka 3.9.1 returns ArrayBuffer instead of ArraySeq
+        final scala.collection.Seq<EndPoint> listeners = kafka.advertisedListeners();
+        final EndPoint endPoint = listeners.apply(0);


Are .apply(0) and .head() equivalent?

Api has been refactored

final EndPoint endPoint = ((ArrayBuffer) kafka.advertisedListeners()).head();

sanath-madhav · 2025-10-30T10:18:54Z

@sanath-madhav Please update the required details in the PR, such as link to ticket it resolves, section for "Describe behaviour before the change", checklist, etc.

Updated the respective details

sanath-madhav · 2025-10-30T10:20:42Z

@sanath-madhav Please update the required details in the PR, such as link to ticket it resolves, section for "Describe behaviour before the change", checklist, etc.

Updated respective details

kaushalaroraharman · 2025-11-03T05:47:35Z

                                    mqttClientId, platform);
                    }
-                });
+                }).join();


This change doesn't seem to be related to kafka streams upgrade. If fixing an existing bug, please do so in a separate PR.

522 - Kafka steram client upgrade initial change

d234bb8

kaushalaroraharman self-requested a review October 29, 2025 06:33

522 - Added changes for kafka upgrade - v1

6d2ac0d

kaushalaroraharman requested changes Oct 30, 2025

View reviewed changes

522 - Fixing all review comments

605f0a5

kaushalaroraharman reviewed Nov 3, 2025

View reviewed changes

kaushalaroraharman approved these changes Nov 3, 2025

View reviewed changes

kaushalaroraharman merged commit d2d0454 into eclipse-ecsp:feature/74-feat-upgrade-kafka-streams-version-to-410 Nov 3, 2025
1 check passed

Conversation

sanath-madhav commented Oct 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe behaviour before the change

Describe behaviour after the change

Pull request checklist

Does this introduce a breaking change?

Uh oh!

kaushalaroraharman commented Oct 30, 2025

Uh oh!

kaushalaroraharman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sanath-madhav commented Oct 30, 2025

Uh oh!

sanath-madhav commented Oct 30, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

sanath-madhav commented Oct 28, 2025 •

edited

Loading