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 1 commit
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
39 changes: 31 additions & 8 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 @@ -266,6 +266,29 @@ object TransactionLog {
}
}
}

/**
* 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 txnMetadata = TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value)

@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.

Should we change readTxnRecordValue to return an Option? Glancing at that code, it looks like another possible NPE in TransactionStateManager#loadTransactions

@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.

Yeah, good suggestion. Pushed a change.

val valueString = if (txnMetadata == null) {
"<DELETE>"
} else {
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
112 changes: 17 additions & 95 deletions core/src/main/scala/kafka/tools/DumpLogSegments.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,18 @@
package kafka.tools

import java.io._
import java.nio.ByteBuffer

import kafka.coordinator.group.{GroupMetadataKey, GroupMetadataManager, OffsetKey}
import kafka.coordinator.group.GroupMetadataManager
import kafka.coordinator.transaction.TransactionLog
import kafka.log._
import kafka.serializer.Decoder
import kafka.utils._
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
import org.apache.kafka.common.KafkaException
import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.utils.Utils

import scala.collection.{Map, mutable}
import scala.collection.mutable.ArrayBuffer
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object DumpLogSegments {

Expand Down Expand Up @@ -212,7 +209,7 @@ object DumpLogSegments {
}
}

private trait MessageParser[K, V] {
private[kafka] trait MessageParser[K, V] {
def parse(record: Record): (Option[K], Option[V])
}

Expand All @@ -233,93 +230,6 @@ object DumpLogSegments {
}
}

private class TransactionLogMessageParser extends MessageParser[String, String] {

override def parse(record: Record): (Option[String], Option[String]) = {
val txnKey = TransactionLog.readTxnRecordKey(record.key)
val txnMetadata = TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value)

val keyString = s"transactionalId=${txnKey.transactionalId}"
val valueString = 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))
}

}

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

private def parseOffsets(offsetKey: OffsetKey, payload: ByteBuffer) = {
val group = offsetKey.key.group
val topicPartition = offsetKey.key.topicPartition
val offset = GroupMetadataManager.readOffsetMessageValue(payload)

val keyString = s"offset::$group:${topicPartition.topic}:${topicPartition.partition}"
val valueString = if (offset.metadata.isEmpty)
String.valueOf(offset.offset)
else
s"${offset.offset}:${offset.metadata}"

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

private def parseGroupMetadata(groupMetadataKey: GroupMetadataKey, payload: ByteBuffer) = {
val groupId = groupMetadataKey.key
val group = GroupMetadataManager.readGroupMessageValue(groupId, payload, Time.SYSTEM)
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 = hex(Utils.toArray(partitionAssignment.userData()))

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

val keyString = Json.encodeAsString(Map("metadata" -> groupId).asJava)

val valueString = Json.encodeAsString(Map(
"protocolType" -> protocolType,
"protocol" -> group.protocolOrNull,
"generationId" -> group.generationId,
"assignment" -> assignment
).asJava)

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

override def parse(record: Record): (Option[String], Option[String]) = {
if (!record.hasValue)
(None, None)
else 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)")
}
}
}
}

/* print out the contents of the log */
private def dumpLog(file: File,
printContents: Boolean,
Expand Down Expand Up @@ -450,6 +360,18 @@ object DumpLogSegments {
}
}

private class OffsetsMessageParser extends MessageParser[String, String] {
override def parse(record: Record): (Option[String], Option[String]) = {
GroupMetadataManager.formatRecordKeyAndValue(record)
}
}

private class TransactionLogMessageParser extends MessageParser[String, String] {
override def parse(record: Record): (Option[String], Option[String]) = {
TransactionLog.formatRecordKeyAndValue(record)
}
}

private class DumpLogSegmentsOptions(args: Array[String]) extends CommandDefaultOptions(args) {
val printOpt = parser.accepts("print-data-log", "if set, printing the messages content when dumping data logs. Automatically set if any decoder option is specified.")
val verifyOpt = parser.accepts("verify-index-only", "if set, just verify the index log without printing its content.")
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ class GroupCoordinatorTest {

val deadGroupId = "deadGroupId"

groupCoordinator.groupManager.addGroup(new GroupMetadata(deadGroupId, Dead, new MockTime()))
groupCoordinator.groupManager.addGroup(new GroupMetadata(deadGroupId, Dead, new MockTime()))
val joinGroupResult = dynamicJoinGroup(deadGroupId, memberId, protocolType, protocols)
assertEquals(Errors.COORDINATOR_NOT_AVAILABLE, joinGroupResult.error)
}
Expand Down
Loading