diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java index a0ac5e83d6d15..d41e214000f36 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java +++ b/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) { diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java b/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java index a82443037102c..77d9d74966476 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java +++ b/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 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); + } } diff --git a/core/src/main/resources/common/message/TransactionLogKey.json b/core/src/main/resources/common/message/TransactionLogKey.json new file mode 100644 index 0000000000000..f6fdbfbf9d625 --- /dev/null +++ b/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"} + ] +} diff --git a/core/src/main/resources/common/message/TransactionLogValue.json b/core/src/main/resources/common/message/TransactionLogValue.json new file mode 100644 index 0000000000000..153dc31253407 --- /dev/null +++ b/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"} + ] +} diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index dd2eeb6f360ca..da1bd4c19edc2 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/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) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala index 7fc1f3b50cc96..d27b704326410 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala @@ -20,13 +20,14 @@ import java.io.PrintStream import java.nio.ByteBuffer import java.nio.charset.StandardCharsets +import kafka.internals.generated.{TransactionLogKey, TransactionLogValue} import org.apache.kafka.clients.consumer.ConsumerRecord -import org.apache.kafka.common.protocol.types.Type._ -import org.apache.kafka.common.protocol.types._ +import org.apache.kafka.common.protocol.{ByteBufferAccessor, MessageUtil} import org.apache.kafka.common.record.{CompressionType, Record, RecordBatch} -import org.apache.kafka.common.{KafkaException, MessageFormatter, TopicPartition} +import org.apache.kafka.common.{MessageFormatter, TopicPartition} import scala.collection.mutable +import scala.jdk.CollectionConverters._ /** * Messages stored for the transaction topic represent the producer id and transactional status of the corresponding @@ -53,91 +54,14 @@ object TransactionLog { val EnforcedCompressionType: CompressionType = CompressionType.NONE val EnforcedRequiredAcks: Short = (-1).toShort - // log message formats - - private object KeySchema { - private val TXN_ID_KEY = "transactional_id" - - private val V0 = new Schema(new Field(TXN_ID_KEY, STRING)) - private val SCHEMAS = Map(0 -> V0) - - val CURRENT_VERSION = 0.toShort - val CURRENT = schemaForKey(CURRENT_VERSION) - - val TXN_ID_FIELD = V0.get(TXN_ID_KEY) - - def ofVersion(version: Int): Option[Schema] = SCHEMAS.get(version) - } - - private object ValueSchema { - private val ProducerIdKey = "producer_id" - private val ProducerEpochKey = "producer_epoch" - private val TxnTimeoutKey = "transaction_timeout" - private val TxnStatusKey = "transaction_status" - private val TxnPartitionsKey = "transaction_partitions" - private val TxnEntryTimestampKey = "transaction_entry_timestamp" - private val TxnStartTimestampKey = "transaction_start_timestamp" - - private val PartitionIdsKey = "partition_ids" - private val TopicKey = "topic" - private val PartitionsSchema = new Schema(new Field(TopicKey, STRING), - new Field(PartitionIdsKey, new ArrayOf(INT32))) - - private val V0 = new Schema(new Field(ProducerIdKey, INT64, "Producer id in use by the transactional id."), - new Field(ProducerEpochKey, INT16, "Epoch associated with the producer id"), - new Field(TxnTimeoutKey, INT32, "Transaction timeout in milliseconds"), - new Field(TxnStatusKey, INT8, - "TransactionState the transaction is in"), - new Field(TxnPartitionsKey, ArrayOf.nullable(PartitionsSchema), - "Set of partitions involved in the transaction"), - new Field(TxnEntryTimestampKey, INT64, "Time the transaction was last updated"), - new Field(TxnStartTimestampKey, INT64, "Time the transaction was started")) - - private val Schemas = Map(0 -> V0) - - val CurrentVersion = 0.toShort - val Current = schemaForValue(CurrentVersion) - - val ProducerIdField = V0.get(ProducerIdKey) - val ProducerEpochField = V0.get(ProducerEpochKey) - val TxnTimeoutField = V0.get(TxnTimeoutKey) - val TxnStatusField = V0.get(TxnStatusKey) - val TxnPartitionsField = V0.get(TxnPartitionsKey) - val TxnEntryTimestampField = V0.get(TxnEntryTimestampKey) - val TxnStartTimestampField = V0.get(TxnStartTimestampKey) - - val PartitionsTopicField = PartitionsSchema.get(TopicKey) - val PartitionIdsField = PartitionsSchema.get(PartitionIdsKey) - - def ofVersion(version: Int): Option[Schema] = Schemas.get(version) - } - - private def schemaForKey(version: Int) = { - KeySchema.ofVersion(version).getOrElse { - throw new KafkaException(s"Unknown transaction log message key schema version $version") - } - } - - private def schemaForValue(version: Int) = { - ValueSchema.ofVersion(version).getOrElse { - throw new KafkaException(s"Unknown transaction log message value schema version $version") - } - } - /** * Generates the bytes for transaction log message key * * @return key bytes */ private[transaction] def keyToBytes(transactionalId: String): Array[Byte] = { - import KeySchema._ - val key = new Struct(CURRENT) - key.set(TXN_ID_FIELD, transactionalId) - - val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf) - byteBuffer.putShort(CURRENT_VERSION) - key.writeTo(byteBuffer) - byteBuffer.array() + MessageUtil.toBytes(TransactionLogKey.HIGHEST_SUPPORTED_VERSION, + new TransactionLogKey().setTransactionalId(transactionalId)) } /** @@ -146,38 +70,27 @@ object TransactionLog { * @return value payload bytes */ private[transaction] def valueToBytes(txnMetadata: TxnTransitMetadata): Array[Byte] = { - import ValueSchema._ - val value = new Struct(Current) - value.set(ProducerIdField, txnMetadata.producerId) - value.set(ProducerEpochField, txnMetadata.producerEpoch) - value.set(TxnTimeoutField, txnMetadata.txnTimeoutMs) - value.set(TxnStatusField, txnMetadata.txnState.byte) - value.set(TxnEntryTimestampField, txnMetadata.txnLastUpdateTimestamp) - value.set(TxnStartTimestampField, txnMetadata.txnStartTimestamp) - - if (txnMetadata.txnState == Empty) { - if (txnMetadata.topicPartitions.nonEmpty) + if (txnMetadata.txnState == Empty && txnMetadata.topicPartitions.nonEmpty) throw new IllegalStateException(s"Transaction is not expected to have any partitions since its state is ${txnMetadata.txnState}: $txnMetadata") - value.set(TxnPartitionsField, null) - } else { - // first group the topic partitions by their topic names - val topicAndPartitions = txnMetadata.topicPartitions.groupBy(_.topic()) - - val partitionArray = topicAndPartitions.map { case(topic, partitions) => - val topicPartitionsStruct = value.instance(TxnPartitionsField) - val partitionIds: Array[Integer] = partitions.map(topicPartition => Integer.valueOf(topicPartition.partition())).toArray - topicPartitionsStruct.set(PartitionsTopicField, topic) - topicPartitionsStruct.set(PartitionIdsField, partitionIds) - topicPartitionsStruct - } - value.set(TxnPartitionsField, partitionArray.toArray) - } - - val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf) - byteBuffer.putShort(CurrentVersion) - value.writeTo(byteBuffer) - byteBuffer.array() + val transactionPartitions = if (txnMetadata.txnState == Empty) null + else txnMetadata.topicPartitions + .groupBy(_.topic) + .map { case (topic, partitions) => + new TransactionLogValue.PartitionsSchema() + .setTopic(topic) + .setPartitionIds(partitions.map(tp => Integer.valueOf(tp.partition)).toList.asJava) + }.toList.asJava + + MessageUtil.toBytes(TransactionLogValue.HIGHEST_SUPPORTED_VERSION, + new TransactionLogValue() + .setProducerId(txnMetadata.producerId) + .setProducerEpoch(txnMetadata.producerEpoch) + .setTransactionTimeoutMs(txnMetadata.txnTimeoutMs) + .setTransactionStatus(txnMetadata.txnState.byte) + .setTransactionLastUpdateTimestampMs(txnMetadata.txnLastUpdateTimestamp) + .setTransactionStartTimestampMs(txnMetadata.txnStartTimestamp) + .setTransactionPartitions(transactionPartitions)) } /** @@ -187,15 +100,13 @@ object TransactionLog { */ def readTxnRecordKey(buffer: ByteBuffer): TxnKey = { val version = buffer.getShort - val keySchema = schemaForKey(version) - val key = keySchema.read(buffer) - - if (version == KeySchema.CURRENT_VERSION) { - val transactionalId = key.getString(KeySchema.TXN_ID_FIELD) - TxnKey(version, transactionalId) - } else { - throw new IllegalStateException(s"Unknown version $version from the transaction log message") - } + if (version >= TransactionLogKey.LOWEST_SUPPORTED_VERSION && version <= TransactionLogKey.HIGHEST_SUPPORTED_VERSION) { + val value = new TransactionLogKey(new ByteBufferAccessor(buffer), version) + TxnKey( + version = version, + transactionalId = value.transactionalId + ) + } else throw new IllegalStateException(s"Unknown version $version from the transaction log message") } /** @@ -204,48 +115,33 @@ object TransactionLog { * @return a transaction metadata object from the message */ def readTxnRecordValue(transactionalId: String, buffer: ByteBuffer): Option[TransactionMetadata] = { - if (buffer == null) { // tombstone - None - } else { - import ValueSchema._ + // tombstone + if (buffer == null) None + else { val version = buffer.getShort - val valueSchema = schemaForValue(version) - val value = valueSchema.read(buffer) - - if (version == CurrentVersion) { - val producerId = value.getLong(ProducerIdField) - val epoch = value.getShort(ProducerEpochField) - val timeout = value.getInt(TxnTimeoutField) - - val stateByte = value.getByte(TxnStatusField) - val state = TransactionMetadata.byteToState(stateByte) - val entryTimestamp = value.getLong(TxnEntryTimestampField) - val startTimestamp = value.getLong(TxnStartTimestampField) - - val transactionMetadata = new TransactionMetadata(transactionalId, producerId, RecordBatch.NO_PRODUCER_ID, - epoch, RecordBatch.NO_PRODUCER_EPOCH, timeout, state, mutable.Set.empty[TopicPartition],startTimestamp, entryTimestamp) - - if (!state.equals(Empty)) { - val topicPartitionArray = value.getArray(TxnPartitionsField) - - topicPartitionArray.foreach { memberMetadataObj => - val memberMetadata = memberMetadataObj.asInstanceOf[Struct] - val topic = memberMetadata.getString(PartitionsTopicField) - val partitionIdArray = memberMetadata.getArray(PartitionIdsField) - - val topicPartitions = partitionIdArray.map { partitionIdObj => - val partitionId = partitionIdObj.asInstanceOf[Integer] - new TopicPartition(topic, partitionId) - } - - transactionMetadata.addPartitions(topicPartitions.toSet) - } - } - + if (version >= TransactionLogValue.LOWEST_SUPPORTED_VERSION && version <= TransactionLogValue.HIGHEST_SUPPORTED_VERSION) { + val value = new TransactionLogValue(new ByteBufferAccessor(buffer), version) + val transactionMetadata = new TransactionMetadata( + transactionalId = transactionalId, + producerId = value.producerId, + lastProducerId = RecordBatch.NO_PRODUCER_ID, + producerEpoch = value.producerEpoch, + lastProducerEpoch = RecordBatch.NO_PRODUCER_EPOCH, + txnTimeoutMs = value.transactionTimeoutMs, + state = TransactionMetadata.byteToState(value.transactionStatus), + topicPartitions = mutable.Set.empty[TopicPartition], + txnStartTimestamp = value.transactionStartTimestampMs, + txnLastUpdateTimestamp = value.transactionLastUpdateTimestampMs) + + if (!transactionMetadata.state.equals(Empty)) + value.transactionPartitions.forEach(partitionsSchema => + transactionMetadata.addPartitions(partitionsSchema.partitionIds + .asScala + .map(partitionId => new TopicPartition(partitionsSchema.topic, partitionId)) + .toSet) + ) Some(transactionMetadata) - } else { - throw new IllegalStateException(s"Unknown version $version from the transaction log message value") - } + } else throw new IllegalStateException(s"Unknown version $version from the transaction log message value") } } @@ -291,5 +187,5 @@ object TransactionLog { } case class TxnKey(version: Short, transactionalId: String) { - override def toString: String = transactionalId.toString + override def toString: String = transactionalId }