kafka 3.8 / 3.9

Author: everesio

proxy/processor_default.go

@@ -159,7 +159,7 @@ func (handler *DefaultRequestHandler) mustReply(requestKeyVersion *protocol.Requ
 				return false, nil, err
 			}
 
-		case 3, 4, 5, 6, 7, 8, 9, 10:
+		case 3, 4, 5, 6, 7, 8, 9, 10, 11:
 			// CorrelationID + ClientID
 			if err = acksReader.ReadAndDiscardHeaderV1Part(reader); err != nil {
 				return false, nil, err

proxy/protocol/request_key_version.go

@@ -360,6 +360,32 @@ func (r *RequestKeyVersion) ResponseHeaderVersion() int16 {
 		return 1
 	case 74: // ListClientMetricsResources
 		return 1
+	case 75: // DescribeTopicPartitions
+		return 1
+	case 76: // ShareGroupHeartbeat
+		return 1
+	case 77: // ShareGroupDescribe
+		return 1
+	case 78: // ShareFetch
+		return 1
+	case 79: // ShareAcknowledge
+		return 1
+	case 80: // AddRaftVoter
+		return 1
+	case 81: // RemoveRaftVoter
+		return 1
+	case 82: // UpdateRaftVoter
+		return 1
+	case 83: // InitializeShareGroupState
+		return 1
+	case 84: // ReadShareGroupState
+		return 1
+	case 85: // WriteShareGroupState
+		return 1
+	case 86: // DeleteShareGroupState
+		return 1
+	case 87: // ReadShareGroupStateSummary
+		return 1
 	default:
 		// throw new UnsupportedVersionException("Unsupported API key " + apiKey);
 		return -1

proxy/protocol/responses.go

@@ -293,7 +293,10 @@ func createFindCoordinatorResponseSchemaVersions() []Schema {
 		&CompactArray{Name: coordinatorsKeyName, Ty: findCoordinatorCoordinatorsSchema4},
 		&SchemaTaggedFields{Name: "response_tagged_fields"},
 	)
-	return []Schema{findCoordinatorResponseV0, findCoordinatorResponseV1, findCoordinatorResponseV2, findCoordinatorResponseV3, findCoordinatorResponseV4}
+	findCoordinatorResponseV5 := findCoordinatorResponseV4
+	findCoordinatorResponseV6 := findCoordinatorResponseV5
+
+	return []Schema{findCoordinatorResponseV0, findCoordinatorResponseV1, findCoordinatorResponseV2, findCoordinatorResponseV3, findCoordinatorResponseV4, findCoordinatorResponseV5, findCoordinatorResponseV6}
 }
 
 func modifyMetadataResponse(decodedStruct *Struct, fn config.NetAddressMappingFunc) error {