KAFKA-14895: [1/N] Move AddPartitionsToTxnManager files to java #19879

brandboat · 2025-06-02T12:13:58Z

Move AddPartitionsToTxnManager to server module and convert to Java.
This patch moves AddPartitionsToTxnManager from the core module to the
server module, with its package updated from kafka.server to
org.apache.kafka.server.transaction. Additionally, several
configuration used by AddPartitionsToTxnManager are moved from
KafkaConfig.scala to AbstractKafkaConfig.java.

brokerId
requestTimeoutMs
controllerListenerNames
interBrokerListenerName
interBrokerSecurityProtocol
effectiveListenerSecurityProtocolMap

The next PR will move AddPartitionsToTxnManagerTest.scala to java

Reviewers: Justine Olshan jolshan@confluent.io, Chia-Ping Tsai
chia7712@gmail.com

brandboat · 2025-06-02T12:17:59Z

The next PR will move AddPartitionsToTxnManagerTest.scala to java

brandboat · 2025-06-02T14:31:54Z

AC, gentle ping @chia7712, @jolshan. Could you please take a look when you are available? Thank you!

chia7712 · 2025-06-02T19:00:34Z

server/src/main/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManager.java

+            // Note: Synchronization is not needed on inflightNodes since it is always accessed from this thread.
+            inflightNodes.remove(node);
+            if (response.authenticationException() != null) {
+                log.error(String.format("AddPartitionsToTxnRequest failed for node %s with an authentication exception.", response.destination()),


log.error("AddPartitionsToTxnRequest failed for node {} with an authentication exception.", response.destination(), response.authenticationException());

chia7712 · 2025-06-02T19:24:20Z

server/src/main/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManager.java

+    @Override
+    public Collection<RequestAndCompletionHandler> generateRequests() {
+        // build and add requests to the queue
+        List<RequestAndCompletionHandler> list = new ArrayList<>();


Please consider using Iterator.remove to optimize it.

public Collection<RequestAndCompletionHandler> generateRequests() { // build and add requests to the queue List<RequestAndCompletionHandler> list = new ArrayList<>(); long currentTimeMs = time.milliseconds(); synchronized (nodesToTransactions) { var iter = nodesToTransactions.entrySet().iterator(); while (iter.hasNext()) { var entry = iter.next(); var node = entry.getKey(); // 第一次 next() var transactionDataAndCallbacks = entry.getValue(); // 第二次 next()，這裡有問題 if (!inflightNodes.contains(node)) { list.add(new RequestAndCompletionHandler( currentTimeMs, node, AddPartitionsToTxnRequest.Builder.forBroker(transactionDataAndCallbacks.transactionData()), new AddPartitionsToTxnHandler(node, transactionDataAndCallbacks) )); inflightNodes.add(node); iter.remove(); } } } return list; }

chia7712 · 2025-06-02T19:28:58Z

server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java

+        }
+    }
+
+    private Map.Entry<ListenerName, SecurityProtocol> getInterBrokerListenerNameAndSecurityProtocol() {


interBrokerListenerNameAndSecurityProtocol

chia7712 · 2025-06-02T19:30:03Z

server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java

+        }
+    }
+
+    private static SecurityProtocol getSecurityProtocol(String protocolName, String configName) {


securityProtocol

chia7712 · 2025-06-02T19:42:02Z

core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala

@@ -275,7 +275,7 @@ abstract class QuorumTestHarness extends Logging {
    formatter.addDirectory(metadataDir.getAbsolutePath)
    formatter.setReleaseVersion(metadataVersion)
    formatter.setUnstableFeatureVersionsEnabled(true)
-    formatter.setControllerListenerName(config.controllerListenerNames.head)
+    formatter.setControllerListenerName(config.controllerListenerNames.stream.findFirst.orElseThrow)


formatter.setControllerListenerName(config.controllerListenerNames.get(0))

I went with this approach at first just to keep things consistent—since .head throws a NoSuchElementException when the list is empty, and stream.findFirst().orElseThrow() behaves the same way. get(0) would throw an IndexOutOfBoundsException instead, so it felt a bit different. But perhaps I'm overthinking.

How about using Objects.requireNonNull?

ummm - you want to have consistent exception - that is not required I think - get(0) is sufficient enough

Thanks for the comment! I've applied the get(0) approach in this PR.

chia7712 · 2025-06-02T19:42:41Z

...n-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java

@@ -137,7 +137,9 @@ public ListenerName controllerListenerName() {
                    .next()
                    .config()
                    .controllerListenerNames()
-                    .head()
+                    .stream()


return new ListenerName( Objects.requireNonNull(controllers() .values() .iterator() .next() .config() .controllerListenerNames() .get(0)) );

jolshan · 2025-06-02T20:02:45Z

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -216,7 +208,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
  def quotaConfig: QuotaConfig = _quotaConfig

  /** ********* General Configuration ***********/
-  var brokerId: Int = getInt(ServerConfigs.BROKER_ID_CONFIG)


Was the only remaining usage of these configs AddPartitionsToTxnManager?

Nope, this config is used in quite a few places, e.g. Partition.scala. But yeah, moving it to AbstractKafkaConfig is totally fine since KafkaConfig.scala extends AbstractKafkaConfig.java.

jolshan · 2025-06-02T20:04:16Z

core/src/main/scala/kafka/tools/StorageTool.scala

@@ -126,7 +126,7 @@ object StorageTool extends Logging {
      setClusterId(namespace.getString("cluster_id")).
      setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled).
      setIgnoreFormatted(namespace.getBoolean("ignore_formatted")).
-      setControllerListenerName(config.controllerListenerNames.head).
+      setControllerListenerName(config.controllerListenerNames.stream.findFirst.orElseThrow).


should we do get(0) here like some of the other suggestions?

jolshan · 2025-06-02T20:26:46Z

I reviewed the main bulk of the change. It looks pretty good! Just some small nits already mentioned. 👍

chia7712 · 2025-06-05T11:04:07Z

> Task :core:compileTestScala
[Error] /home/chia7712/project/kafka/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:2450:32: type mismatch;
 found   : scala.collection.Seq[org.apache.kafka.common.TopicPartition]
 required: java.util.Collection[org.apache.kafka.common.TopicPartition]
[Error] /home/chia7712/project/kafka/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:2460:15: org.apache.kafka.server.transaction.AddPartitionsToTxnManager.AppendCallback does not take parameters
two errors found

@brandboat please fix the build error

chia7712 · 2025-06-05T15:14:39Z

server/src/main/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManager.java

+     * An interface which handles the Partition Response based on the Request Version and the exact operation.
+     */
+    @FunctionalInterface
+    public interface TransactionSupportedOperation {


Could you please consider using java enum instead?

/** * handles the Partition Response based on the Request Version and the exact operation. */ enum TransactionSupportedOperation { /** * This is the default workflow which maps to cases when the Produce Request Version or the * Txn_offset_commit request was lower than the first version supporting the new Error Class. */ DEFAULT_ERROR(false), /** * This maps to the case when the clients are updated to handle the TransactionAbortableException. */ GENERIC_ERROR_SUPPORTED(false), /** * This allows the partition to be added to the transactions inflight with the Produce and TxnOffsetCommit requests. * Plus the behaviors in genericErrorSupported. */ ADD_PARTITION(true); private final boolean supportsEpochBump; TransactionSupportedOperation(boolean supportsEpochBump) { this.supportsEpochBump = supportsEpochBump; } }

ah, good call.

KAKFA-14895: [1/N] Move AddPartitionsToTxnManager files to java

f5c0269

github-actions bot added the core label Jun 2, 2025

brandboat changed the title ~~KAKFA-14895: (WIP) [1/N] Move AddPartitionsToTxnManager files to java~~ KAFKA-14895: (WIP) [1/N] Move AddPartitionsToTxnManager files to java Jun 2, 2025

brandboat changed the title ~~KAFKA-14895: (WIP) [1/N] Move AddPartitionsToTxnManager files to java~~ KAFKA-14895: [1/N] Move AddPartitionsToTxnManager files to java Jun 2, 2025

chia7712 reviewed Jun 2, 2025

View reviewed changes

jolshan reviewed Jun 2, 2025

View reviewed changes

brandboat added 3 commits June 3, 2025 13:28

Apply .get(0)

837923e

Address comments

fe72850

Merge remote-tracking branch 'origin/trunk' into KAFKA-14895

bc7b0ad

brandboat added 2 commits June 5, 2025 20:18

Merge remote-tracking branch 'origin/trunk' into KAFKA-14895

ce00a0d

Fix conflicts

996bc4a

chia7712 reviewed Jun 5, 2025

View reviewed changes

brandboat added 2 commits June 6, 2025 00:13

Address comments

ecd4692

Merge remote-tracking branch 'origin/trunk' into KAFKA-14895

d48902f

chia7712 approved these changes Jun 7, 2025

View reviewed changes

chia7712 merged commit 83fb40d into apache:trunk Jun 7, 2025
21 of 23 checks passed

brandboat deleted the KAFKA-14895 branch June 7, 2025 16:54

brandboat restored the KAFKA-14895 branch June 7, 2025 16:54

brandboat deleted the KAFKA-14895 branch June 7, 2025 16:54

KAFKA-14895: [1/N] Move AddPartitionsToTxnManager files to java #19879

KAFKA-14895: [1/N] Move AddPartitionsToTxnManager files to java #19879

Uh oh!

Conversation

brandboat commented Jun 2, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

brandboat commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

brandboat commented Jun 2, 2025

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!

jolshan commented Jun 2, 2025

Uh oh!

chia7712 commented Jun 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

brandboat commented Jun 2, 2025 •

edited by github-actions bot

Loading

brandboat commented Jun 2, 2025 •

edited

Loading