KAFKA-20738: Producer 2PC InitProducerId request must negotiate v6

clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdRequest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.InitProducerIdRequestData;
 import org.apache.kafka.common.message.InitProducerIdResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -28,12 +29,18 @@ public static class Builder extends AbstractRequest.Builder<InitProducerIdReques
         public final InitProducerIdRequestData data;
 
         public Builder(InitProducerIdRequestData data) {
-            super(ApiKeys.INIT_PRODUCER_ID);
+            // enable2Pc/keepPreparedTxn are v6-only (KIP-939); v6 is the unstable latest
+            // version, so only allow it when a 2PC field is set.
+            super(ApiKeys.INIT_PRODUCER_ID, data.enable2Pc() || data.keepPreparedTxn());
             this.data = data;
         }
 
         @Override
         public InitProducerIdRequest build(short version) {
+            if (version < 6 && (data.enable2Pc() || data.keepPreparedTxn()))
+                throw new UnsupportedVersionException("Cannot use 2PC (enable2Pc/keepPreparedTxn) with InitProducerId v"
+                    + version + "; the broker does not support InitProducerId v6 (KIP-939).");
+
             if (data.transactionTimeoutMs() <= 0)
                 throw new IllegalArgumentException("transaction timeout value is not positive: " + data.transactionTimeoutMs());
 

clients/src/test/java/org/apache/kafka/common/requests/InitProducerIdRequestTest.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.InitProducerIdRequestData;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class InitProducerIdRequestTest {
+
+    private InitProducerIdRequestData baseData() {
+        return new InitProducerIdRequestData()
+            .setTransactionalId("txn-2pc")
+            .setTransactionTimeoutMs(10);
+    }
+
+    @Test
+    public void testBuilderAllowsV6WhenEnable2PcSet() {
+        // enable2Pc is a v6-only field (KIP-939); the builder must allow v6 so
+        // NetworkClient can negotiate it against a 2PC-capable broker.
+        InitProducerIdRequest.Builder builder =
+            new InitProducerIdRequest.Builder(baseData().setEnable2Pc(true));
+        assertEquals((short) 6, builder.latestAllowedVersion());
+    }
+
+    @Test
+    public void testBuilderAllowsV6WhenKeepPreparedTxnSet() {
+        InitProducerIdRequest.Builder builder =
+            new InitProducerIdRequest.Builder(baseData().setKeepPreparedTxn(true));
+        assertEquals((short) 6, builder.latestAllowedVersion());
+    }
+
+    @Test
+    public void testBuilderCapsAtStableVersionWithout2Pc() {
+        // Control: a plain producer keeps the stable cap, unaffected by the fix.
+        InitProducerIdRequest.Builder builder = new InitProducerIdRequest.Builder(baseData());
+        assertEquals((short) 5, builder.latestAllowedVersion());
+    }
+}