Upgrade to kafka-clients 3.4.0

encmod/src/main/java/io/strimzi/kafka/topicenc/EncryptionModule.java

@@ -11,7 +11,7 @@
 
 import javax.crypto.SecretKey;
 
-import org.apache.kafka.common.message.FetchResponseData.FetchablePartitionResponse;
+import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.FetchResponseData.FetchableTopicResponse;
 import org.apache.kafka.common.message.ProduceRequestData.PartitionProduceData;
 import org.apache.kafka.common.message.ProduceRequestData.TopicProduceData;
@@ -112,18 +112,18 @@ public boolean decrypt(FetchableTopicResponse fetchRsp)
 
         // If this far, the data was encrypted.
         // Navigate into each record and decrypt.
-        for (FetchablePartitionResponse partitionData : fetchRsp.partitionResponses()) {
+        for (FetchResponseData.PartitionData partitionData : fetchRsp.partitions()) {
 
             if (LOGGER.isDebugEnabled()) {
                 String msg = String.format(
                         "partition: %d, logStartOffset: %08X, lastStableOffset: %08X, "
                                 + "partition leader epoch: %04X",
-                        partitionData.partition(), partitionData.currentLeader().leaderEpoch(),
+                        partitionData.partitionIndex(), partitionData.currentLeader().leaderEpoch(),
                         partitionData.logStartOffset(), partitionData.lastStableOffset());
                 LOGGER.debug(msg);
             }
 
-            MemoryRecords recs = (MemoryRecords) partitionData.recordSet();
+            MemoryRecords recs = (MemoryRecords) partitionData.records();
 
             long firstOffset = getFirstOffset(recs);
             MemoryRecordsBuilder builder = createMemoryRecsBuilder(recs.sizeInBytes(),
@@ -147,7 +147,7 @@ public boolean decrypt(FetchableTopicResponse fetchRsp)
             }
             // overwrite the partition's memoryrecords with the decrypted records:
             MemoryRecords newRecs = builder.build();
-            partitionData.setRecordSet(newRecs);
+            partitionData.setRecords(newRecs);
         }
         return true;
     }

pom.xml

@@ -72,7 +72,7 @@
         <junit-platform-surefire-provider.version>1.3.2</junit-platform-surefire-provider.version>
         <testcontainers.version>1.17.2</testcontainers.version>
 
-        <kafka.version>2.8.0</kafka.version>
+        <kafka.version>3.4.0</kafka.version>
 
         <spotbugs.version>4.7.1</spotbugs.version>
         <fasterxml.jackson-core.version>2.13.3</fasterxml.jackson-core.version>

vertx-proxy/src/main/java/io/strimzi/kafka/proxy/vertx/MessageHandler.java

@@ -366,7 +366,7 @@ public Buffer processFetchResponse(Buffer buffer, RequestHeader reqHeader)
             throws EncSerDerException, GeneralSecurityException, KmsException {
         // instantiate FetchResponse instance
         KafkaRspMsg rsp = new KafkaRspMsg(buffer, reqHeader.apiVersion());
-        FetchResponse<?> fetch = (FetchResponse<?>) AbstractResponse.parseResponse(rsp.getPayload(),
+        FetchResponse fetch = (FetchResponse) AbstractResponse.parseResponse(rsp.getPayload(),
                 reqHeader);
 
         // iterate through response records, decrypting where needed

vertx-proxy/src/main/java/io/strimzi/kafka/proxy/vertx/msg/MsgUtil.java

@@ -106,7 +106,7 @@ public static Buffer toSendBuffer(byte[] header, AbstractRequest req) {
      * @param req
      * @return
      */
-    public static Buffer toSendBuffer(FetchResponse<?> fetchRsp, RequestHeader reqHeader) {
+    public static Buffer toSendBuffer(FetchResponse fetchRsp, RequestHeader reqHeader) {
         ByteBuffer serializedRsp = serialize(fetchRsp, reqHeader);
         byte[] rspBytes = serializedRsp.array();
         int bufLen = Integer.BYTES + rspBytes.length;
@@ -124,7 +124,7 @@ public static Buffer toSendBuffer(FetchResponse<?> fetchRsp, RequestHeader reqHe
      * @param fetchRsp
      * @return
      */
-    private static ByteBuffer serialize(FetchResponse<?> fetchRsp, RequestHeader reqHeader) {
+    private static ByteBuffer serialize(FetchResponse fetchRsp, RequestHeader reqHeader) {
         ResponseHeader rspHeader = reqHeader.toResponseHeader();
         //System.out.println("****** req: " + reqHeader.apiVersion() + " rsp: " + rspHeader.headerVersion());
         return RequestUtils.serialize(rspHeader.data(), rspHeader.headerVersion(), fetchRsp.data(), reqHeader.apiVersion());

vertx-proxy/src/test/java/io/strimzi/kafka/proxy/vertx/EncModTest.java

@@ -90,7 +90,7 @@ private void testDecryption(File rspMsgFile)
 
         // instantiate the decrypted fetch response
         KafkaRspMsg rsp = new KafkaRspMsg(fetchRspBuf, reqHeader.apiVersion());
-        FetchResponse<?> fetch = (FetchResponse<?>) AbstractResponse.parseResponse(rsp.getPayload(),
+        FetchResponse fetch = (FetchResponse) AbstractResponse.parseResponse(rsp.getPayload(),
                 reqHeader);
 
         FetchResponseData data = fetch.data();
@@ -102,8 +102,8 @@ private void navigate(FetchResponseData data) {
         // This tests the integrity of the decrypted response.
         List<FetchableTopicResponse> responses = data.responses();
         for (FetchableTopicResponse topicRsp : responses) {
-            topicRsp.partitionResponses().forEach(pd -> {
-                MemoryRecords recs = (MemoryRecords) pd.recordSet();
+            topicRsp.partitions().forEach(pd -> {
+                MemoryRecords recs = (MemoryRecords) pd.records();
                 recs.records().forEach(r -> {
                     if (r.hasValue()) {
                         byte[] recordData = new byte[r.valueSize()];