Skip to content

KAFKA-9293; Fix NPE in DumpLogSegments offsets parser and display tombstone keys #7820

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 2 commits into from
Jan 3, 2020
Merged

KAFKA-9293; Fix NPE in DumpLogSegments offsets parser and display tombstone keys #7820

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
77a3852
KAFKA-9293; Fix NPE in DumpLogSegments group/transaction metadata par…
hachikuji Dec 11, 2019
67c3efe
Use Option in result of `readTxnRecordValue`
hachikuji Jan 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 88 additions & 12 deletions core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,18 +34,18 @@ import kafka.utils.CoreUtils.inLock
import kafka.utils._
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.metrics.stats.Meter
import org.apache.kafka.common.metrics.stats.{Avg, Max}
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.metrics.stats.{Avg, Max, Meter}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.protocol.types.Type._
import org.apache.kafka.common.protocol.types._
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.{OffsetCommitRequest, OffsetFetchResponse}
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.requests.{OffsetCommitRequest, OffsetFetchResponse}
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.{KafkaException, TopicPartition}

import scala.collection.JavaConverters._
import scala.collection._
Expand Down Expand Up @@ -1146,8 +1146,7 @@ object GroupMetadataManager {
*
* @return key for offset commit message
*/
private[group] def offsetCommitKey(group: String,
topicPartition: TopicPartition): Array[Byte] = {
def offsetCommitKey(group: String, topicPartition: TopicPartition): Array[Byte] = {
val key = new Struct(CURRENT_OFFSET_KEY_SCHEMA)
key.set(OFFSET_KEY_GROUP_FIELD, group)
key.set(OFFSET_KEY_TOPIC_FIELD, topicPartition.topic)
Expand All @@ -1164,7 +1163,7 @@ object GroupMetadataManager {
*
* @return key bytes for group metadata message
*/
private[group] def groupMetadataKey(group: String): Array[Byte] = {
def groupMetadataKey(group: String): Array[Byte] = {
val key = new Struct(CURRENT_GROUP_KEY_SCHEMA)
key.set(GROUP_KEY_GROUP_FIELD, group)

Expand All @@ -1181,8 +1180,8 @@ object GroupMetadataManager {
* @param apiVersion the api version
* @return payload for offset commit message
*/
private[group] def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata,
apiVersion: ApiVersion): Array[Byte] = {
def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata,
apiVersion: ApiVersion): Array[Byte] = {
// generate commit value according to schema version
val (version, value) = {
if (apiVersion < KAFKA_2_1_IV0 || offsetAndMetadata.expireTimestamp.nonEmpty) {
Expand Down Expand Up @@ -1226,9 +1225,9 @@ object GroupMetadataManager {
* @param apiVersion the api version
* @return payload for offset commit message
*/
private[group] def groupMetadataValue(groupMetadata: GroupMetadata,
assignment: Map[String, Array[Byte]],
apiVersion: ApiVersion): Array[Byte] = {
def groupMetadataValue(groupMetadata: GroupMetadata,
assignment: Map[String, Array[Byte]],
apiVersion: ApiVersion): Array[Byte] = {

val (version, value) = {
if (apiVersion < KAFKA_0_10_1_IV0)
Expand Down Expand Up @@ -1467,6 +1466,83 @@ object GroupMetadataManager {
}
}

/**
* Exposed for printing records using [[kafka.tools.DumpLogSegments]]
*/
def formatRecordKeyAndValue(record: Record): (Option[String], Option[String]) = {

@mumrah mumrah Jan 2, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can one or the other string be missing? Or will they always be present or absent together? Just wondering if this should be Option[Pair[String, String]] or something

@hachikuji hachikuji Jan 2, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree the API is a bit awkward, but I think both cases are possible. For example, a tombstone will have a key, but not a value.

if (!record.hasKey) {
throw new KafkaException("Failed to decode message using offset topic decoder (message had a missing key)")
} else {
GroupMetadataManager.readMessageKey(record.key) match {
case offsetKey: OffsetKey => parseOffsets(offsetKey, record.value)
case groupMetadataKey: GroupMetadataKey => parseGroupMetadata(groupMetadataKey, record.value)
case _ => throw new KafkaException("Failed to decode message using offset topic decoder (message had an invalid key)")
}
}
}

private def parseOffsets(offsetKey: OffsetKey, payload: ByteBuffer): (Option[String], Option[String]) = {
val groupId = offsetKey.key.group
val topicPartition = offsetKey.key.topicPartition
val keyString = s"offset_commit::group=$groupId,partition=$topicPartition"

val offset = GroupMetadataManager.readOffsetMessageValue(payload)
val valueString = if (offset == null) {
"<DELETE>"
} else {
if (offset.metadata.isEmpty)
s"offset=${offset.offset}"
else
s"offset=${offset.offset},metadata=${offset.metadata}"
}

(Some(keyString), Some(valueString))
}

private def parseGroupMetadata(groupMetadataKey: GroupMetadataKey, payload: ByteBuffer): (Option[String], Option[String]) = {
val groupId = groupMetadataKey.key
val keyString = s"group_metadata::group=$groupId"

val group = GroupMetadataManager.readGroupMessageValue(groupId, payload, Time.SYSTEM)
val valueString = if (group == null)
"<DELETE>"
else {
val protocolType = group.protocolType.getOrElse("")

val assignment = group.allMemberMetadata.map { member =>
if (protocolType == ConsumerProtocol.PROTOCOL_TYPE) {
val partitionAssignment = ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(member.assignment))
val userData = Option(partitionAssignment.userData)
.map(Utils.toArray)
.map(hex)
.getOrElse("")

if (userData.isEmpty)
s"${member.memberId}=${partitionAssignment.partitions()}"
else
s"${member.memberId}=${partitionAssignment.partitions()}:$userData"
} else {
s"${member.memberId}=${hex(member.assignment)}"
}
}.mkString("{", ",", "}")

Json.encodeAsString(Map(
"protocolType" -> protocolType,
"protocol" -> group.protocolOrNull,
"generationId" -> group.generationId,
"assignment" -> assignment
).asJava)
}
(Some(keyString), Some(valueString))
}

private def hex(bytes: Array[Byte]): String = {
if (bytes.isEmpty)
""
else
"%X".format(BigInt(1, bytes))
}

}

case class GroupTopicPartition(group: String, topicPartition: TopicPartition) {
Expand Down
53 changes: 38 additions & 15 deletions core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,16 @@
*/
package kafka.coordinator.transaction

import kafka.common.MessageFormatter
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.common.protocol.types.Type._
import org.apache.kafka.common.protocol.types._
import java.io.PrintStream
import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets

import org.apache.kafka.common.record.{CompressionType, RecordBatch}
import kafka.common.MessageFormatter
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.record.{CompressionType, Record, RecordBatch}
import org.apache.kafka.common.{KafkaException, TopicPartition}

import scala.collection.mutable

Expand Down Expand Up @@ -130,7 +130,7 @@ object TransactionLog {
*
* @return key bytes
*/
private[coordinator] def keyToBytes(transactionalId: String): Array[Byte] = {
private[transaction] def keyToBytes(transactionalId: String): Array[Byte] = {
import KeySchema._
val key = new Struct(CURRENT)
key.set(TXN_ID_FIELD, transactionalId)
Expand All @@ -146,7 +146,7 @@ object TransactionLog {
*
* @return value payload bytes
*/
private[coordinator] def valueToBytes(txnMetadata: TxnTransitMetadata): Array[Byte] = {
private[transaction] def valueToBytes(txnMetadata: TxnTransitMetadata): Array[Byte] = {
import ValueSchema._
val value = new Struct(Current)
value.set(ProducerIdField, txnMetadata.producerId)
Expand Down Expand Up @@ -204,9 +204,9 @@ object TransactionLog {
*
* @return a transaction metadata object from the message
*/
def readTxnRecordValue(transactionalId: String, buffer: ByteBuffer): TransactionMetadata = {
def readTxnRecordValue(transactionalId: String, buffer: ByteBuffer): Option[TransactionMetadata] = {
if (buffer == null) { // tombstone
null
None
} else {
import ValueSchema._
val version = buffer.getShort
Expand Down Expand Up @@ -243,7 +243,7 @@ object TransactionLog {
}
}

transactionMetadata
Some(transactionMetadata)
} else {
throw new IllegalStateException(s"Unknown version $version from the transaction log message value")
}
Expand All @@ -256,16 +256,39 @@ object TransactionLog {
Option(consumerRecord.key).map(key => readTxnRecordKey(ByteBuffer.wrap(key))).foreach { txnKey =>
val transactionalId = txnKey.transactionalId
val value = consumerRecord.value
val producerIdMetadata =
if (value == null) "NULL"
else readTxnRecordValue(transactionalId, ByteBuffer.wrap(value))
val producerIdMetadata = if (value == null)
None
else
readTxnRecordValue(transactionalId, ByteBuffer.wrap(value))
output.write(transactionalId.getBytes(StandardCharsets.UTF_8))
output.write("::".getBytes(StandardCharsets.UTF_8))
output.write(producerIdMetadata.toString.getBytes(StandardCharsets.UTF_8))
output.write(producerIdMetadata.getOrElse("NULL").toString.getBytes(StandardCharsets.UTF_8))
output.write("\n".getBytes(StandardCharsets.UTF_8))
}
}
}

/**
* Exposed for printing records using [[kafka.tools.DumpLogSegments]]
*/
def formatRecordKeyAndValue(record: Record): (Option[String], Option[String]) = {
val txnKey = TransactionLog.readTxnRecordKey(record.key)
val keyString = s"transaction_metadata::transactionalId=${txnKey.transactionalId}"

val valueString = TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value) match {
case None => "<DELETE>"

case Some(txnMetadata) => s"producerId:${txnMetadata.producerId}," +
s"producerEpoch:${txnMetadata.producerEpoch}," +
s"state=${txnMetadata.state}," +
s"partitions=${txnMetadata.topicPartitions}," +
s"txnLastUpdateTimestamp=${txnMetadata.txnLastUpdateTimestamp}," +
s"txnTimeoutMs=${txnMetadata.txnTimeoutMs}"
}

(Some(keyString), Some(valueString))
}

}

case class TxnKey(version: Short, transactionalId: String) {
Expand Down
10 changes: 5 additions & 5 deletions core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -347,11 +347,11 @@ class TransactionStateManager(brokerId: Int,
val txnKey = TransactionLog.readTxnRecordKey(record.key)
// load transaction metadata along with transaction state
val transactionalId = txnKey.transactionalId
if (!record.hasValue) {
loadedTransactions.remove(transactionalId)
} else {
val txnMetadata = TransactionLog.readTxnRecordValue(transactionalId, record.value)
loadedTransactions.put(transactionalId, txnMetadata)
TransactionLog.readTxnRecordValue(transactionalId, record.value) match {
case None =>
loadedTransactions.remove(transactionalId)
case Some(txnMetadata) =>
loadedTransactions.put(transactionalId, txnMetadata)
}
currOffset = batch.nextOffset
}
Expand Down
Loading