KAFKA-10736 Convert transaction coordinator metadata schemas to use g…

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java

@@ -81,7 +81,7 @@ public static ByteBuffer serializeSubscription(final Subscription subscription,
                 .setPartitions(topicEntry.getValue()));
         }
 
-        return MessageUtil.serializeMessage(version, data);
+        return MessageUtil.toByteBuffer(version, data);
     }
 
     public static Subscription deserializeSubscription(final ByteBuffer buffer, short version) {
@@ -127,7 +127,7 @@ public static ByteBuffer serializeAssignment(final Assignment assignment, short
                 .setPartitions(topicEntry.getValue()));
         }
 
-        return MessageUtil.serializeMessage(version, data);
+        return MessageUtil.toByteBuffer(version, data);
     }
 
     public static Assignment deserializeAssignment(final ByteBuffer buffer, short version) {

clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java

@@ -19,6 +19,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.kafka.common.protocol.types.RawTaggedField;
+import org.apache.kafka.common.utils.Utils;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -181,7 +182,7 @@ public static boolean compareRawTaggedFields(List<RawTaggedField> first,
         }
     }
 
-    public static ByteBuffer serializeMessage(final short version, final Message message) {
+    public static ByteBuffer toByteBuffer(final short version, final Message message) {
         ObjectSerializationCache cache = new ObjectSerializationCache();
         int size = message.size(cache, version);
         ByteBuffer bytes = ByteBuffer.allocate(2 + size);
@@ -191,4 +192,14 @@ public static ByteBuffer serializeMessage(final short version, final Message mes
         bytes.flip();
         return bytes;
     }
+
+    public static byte[] toBytes(final short version, final Message message) {
+        ByteBuffer buffer = toByteBuffer(version, message);
+        // take the inner array directly if it is full with data
+        if (buffer.hasArray() &&
+                buffer.arrayOffset() == 0 &&
+                buffer.position() == 0 &&
+                buffer.limit() == buffer.array().length) return buffer.array();
+        else return Utils.toArray(buffer);
+    }
 }

core/src/main/resources/common/message/TransactionLogKey.json

@@ -0,0 +1,23 @@
+// 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.
+
+{
+  "type": "data",
+  "name": "TransactionLogKey",
+  "validVersions": "0",
+  "fields": [
+    { "name": "TransactionalId", "type": "string", "versions": "0"}
+  ]
+}

core/src/main/resources/common/message/TransactionLogValue.json

@@ -0,0 +1,38 @@
+// 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.
+
+{
+  "type": "data",
+  "name": "TransactionLogValue",
+  "validVersions": "0",
+  "fields": [
+    { "name": "ProducerId", "type": "int64", "versions": "0",
+      "about": "Producer id in use by the transactional id"},
+    { "name": "ProducerEpoch", "type": "int16", "versions": "0",
+      "about": "Epoch associated with the producer id"},
+    { "name": "TransactionTimeoutMs", "type": "int32", "versions": "0",
+      "about": "Transaction timeout in milliseconds"},
+    { "name": "TransactionStatus", "type": "int8", "versions": "0",
+      "about": "TransactionState the transaction is in"},
+    { "name": "TransactionPartitions", "type": "[]PartitionsSchema", "versions": "0", "nullableVersions": "0",
+      "about": "Set of partitions involved in the transaction", "fields": [
+      { "name": "Topic", "type": "string", "versions": "0"},
+      { "name": "PartitionIds", "type": "[]int32", "versions": "0"}]},
+    { "name": "TransactionLastUpdateTimestampMs", "type": "int64", "versions": "0",
+      "about": "Time the transaction was last updated"},
+    { "name": "TransactionStartTimestampMs", "type": "int64", "versions": "0",
+      "about": "Time the transaction was started"}
+  ]
+}

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -28,20 +28,20 @@ import java.util.concurrent.locks.ReentrantLock
 import com.yammer.metrics.core.Gauge
 import kafka.api.{ApiVersion, KAFKA_0_10_1_IV0, KAFKA_2_1_IV0, KAFKA_2_1_IV1, KAFKA_2_3_IV0}
 import kafka.common.OffsetAndMetadata
-import kafka.internals.generated.{GroupMetadataKey => GroupMetadataKeyData, GroupMetadataValue, OffsetCommitKey, OffsetCommitValue}
+import kafka.internals.generated.{GroupMetadataValue, OffsetCommitKey, OffsetCommitValue, GroupMetadataKey => GroupMetadataKeyData}
 import kafka.log.AppendOrigin
 import kafka.metrics.KafkaMetricsGroup
 import kafka.server.{FetchLogEnd, ReplicaManager}
 import kafka.utils.CoreUtils.inLock
-import kafka.utils._
 import kafka.utils.Implicits._
+import kafka.utils._
 import kafka.zk.KafkaZkClient
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
 import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.metrics.stats.{Avg, Max, Meter}
-import org.apache.kafka.common.protocol.{ByteBufferAccessor, Errors, Message, MessageUtil}
+import org.apache.kafka.common.protocol.{ByteBufferAccessor, Errors, MessageUtil}
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.requests.OffsetFetchResponse.PartitionData
 import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
@@ -996,17 +996,6 @@ object GroupMetadataManager {
   val MetricsGroup: String = "group-coordinator-metrics"
   val LoadTimeSensor: String = "GroupPartitionLoadTime"
 
-  private def serializeMessage(version: Short, message: Message): Array[Byte] = {
-    val buffer = MessageUtil.serializeMessage(version, message)
-    // take the inner array directly if it is full with data
-    if (buffer.hasArray &&
-      buffer.arrayOffset() == 0 &&
-      buffer.position() == 0 &&
-      buffer.limit() == buffer.array().length)
-      buffer.array()
-    else Utils.toArray(buffer)
-  }
-
   /**
    * Generates the key for offset commit message for given (group, topic, partition)
    *
@@ -1015,7 +1004,7 @@ object GroupMetadataManager {
    * @return key for offset commit message
    */
   def offsetCommitKey(groupId: String, topicPartition: TopicPartition): Array[Byte] = {
-    serializeMessage(OffsetCommitKey.HIGHEST_SUPPORTED_VERSION,
+    MessageUtil.toBytes(OffsetCommitKey.HIGHEST_SUPPORTED_VERSION,
       new OffsetCommitKey()
         .setGroup(groupId)
         .setTopic(topicPartition.topic)
@@ -1029,7 +1018,7 @@ object GroupMetadataManager {
    * @return key bytes for group metadata message
    */
   def groupMetadataKey(groupId: String): Array[Byte] = {
-    serializeMessage(GroupMetadataKeyData.HIGHEST_SUPPORTED_VERSION,
+    MessageUtil.toBytes(GroupMetadataKeyData.HIGHEST_SUPPORTED_VERSION,
       new GroupMetadataKeyData()
         .setGroup(groupId))
   }
@@ -1047,7 +1036,7 @@ object GroupMetadataManager {
       if (apiVersion < KAFKA_2_1_IV0 || offsetAndMetadata.expireTimestamp.nonEmpty) 1.toShort
       else if (apiVersion < KAFKA_2_1_IV1) 2.toShort
       else 3.toShort
-    serializeMessage(version, new OffsetCommitValue()
+    MessageUtil.toBytes(version, new OffsetCommitValue()
       .setOffset(offsetAndMetadata.offset)
       .setMetadata(offsetAndMetadata.metadata)
       .setCommitTimestamp(offsetAndMetadata.commitTimestamp)
@@ -1076,7 +1065,7 @@ object GroupMetadataManager {
       else if (apiVersion < KAFKA_2_3_IV0) 2.toShort
       else 3.toShort
 
-    serializeMessage(version, new GroupMetadataValue()
+    MessageUtil.toBytes(version, new GroupMetadataValue()
       .setProtocolType(groupMetadata.protocolType.getOrElse(""))
       .setGeneration(groupMetadata.generationId)
       .setProtocol(groupMetadata.protocolName.orNull)