kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Mohanraj Nagasamy <techie.mohan...@gmail.com>
Subject Re: Feedback: Print schemaId using bin/kafka-dump-log.sh
Date Wed, 08 Jul 2020 22:48:36 GMT
Thanks Adam for providing the feedback! 

On 7/7/20, 5:05 AM, "Adam Bellemare" <adam.bellemare@gmail.com> wrote:

    Hi Mohanraj

    While I see the usefulness of your suggestion, the main issue is that
    you're using the Confluent schema registry's conventions and hardwiring
    them into Kafka core. Given that Confluent's standards are not part of
    Kafka's official standards, I do not think you will get approval to submit
    this code into Kafka core.

    There may be Confluent tools that are available that already do this, or
    perhaps they have their own custom tools available where this may be more
    suitable for submission.

    Adam



    On Mon, Jul 6, 2020 at 11:00 AM Mohanraj Nagasamy <techie.mohanraj@gmail.com>
    wrote:

    > Do anyone have feedback on this? ☺
    >
    > From: Mohanraj Nagasamy <techie.mohanraj@gmail.com>
    > Date: Wednesday, July 1, 2020 at 6:29 PM
    > To: "dev@kafka.apache.org" <dev@kafka.apache.org>
    > Subject: Feedback: Print schemaId using bin/kafka-dump-log.sh
    >
    > When I try to dump kafka logs for diagnosing or debugging a problem, It's
    > handy to see if the kafka log message schemaId or not. If it has got, print
    > the schemaId.
    >
    > I'm soliciting feedback as to whether it is worth making this change to
    > kafka-core codebase.
    >
    > I'm new to the kafka-community - forgive me if this wasn't part of the
    > process.
    >
    > This is the change I made:
    >
    > ```
    >  core/src/main/scala/kafka/tools/DumpLogSegments.scala | 21
    > +++++++++++++++++++--
    >  1 file changed, 19 insertions(+), 2 deletions(-)
    >
    > diff --git a/core/src/main/scala/kafka/tools/DumpLogSegments.scala
    > b/core/src/main/scala/kafka/tools/DumpLogSegments.scala
    > index 9e9546a92..a8750ac3d 100755
    > --- a/core/src/main/scala/kafka/tools/DumpLogSegments.scala
    > +++ b/core/src/main/scala/kafka/tools/DumpLogSegments.scala
    > @@ -35,6 +35,7 @@ object DumpLogSegments {
    >
    >    // visible for testing
    >    private[tools] val RecordIndent = "|"
    > +  private val MAGIC_BYTE = 0x0
    >
    >    def main(args: Array[String]): Unit = {
    >      val opts = new DumpLogSegmentsOptions(args)
    > @@ -277,8 +278,24 @@ object DumpLogSegments {
    >                }
    >              } else if (printContents) {
    >                val (key, payload) = parser.parse(record)
    > -              key.foreach(key => print(s" key: $key"))
    > -              payload.foreach(payload => print(s" payload: $payload"))
    > +              key.foreach(key => {
    > +                val keyBuffer = record.key()
    > +                if (keyBuffer.get() == MAGIC_BYTE) {
    > +                  print(s" keySchemaId: ${keyBuffer.getInt} key: $key")
    > +                }
    > +                else {
    > +                  print(s" key: $key")
    > +                }
    > +              })
    > +
    > +              payload.foreach(payload => {
    > +                val valueBuffer = record.value()
    > +                if (valueBuffer.get() == MAGIC_BYTE) {
    > +                  print(s" payloadSchemaId: ${valueBuffer.getInt}
    > payload: $payload")
    > +                } else {
    > +                  print(s" payload: $payload")
    > +                }
    > +              })
    >              }
    >              println()
    >            }
    > (END)
    > ```
    >
    > And this is how the output looks like:
    >
    > ```
    > $ bin/kafka-dump-log.sh --files
    > data/kafka/logdir/avro_topic_test-0/00000000000000000000.log
    > --print-data-log
    >
    > | offset: 50 CreateTime: 1593570942959 keysize: -1 valuesize: 16 sequence:
    > -1 headerKeys: [] payloadSchemaId: 1 payload:
    > TracRowe
    > baseOffset: 51 lastOffset: 51 count: 1 baseSequence: -1 lastSequence: -1
    > producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional:
    > false isControl: false position: 2918 CreateTime: 1593570958044 size: 101
    > magic: 2 compresscodec: NONE crc: 1913155179 isvalid: true
    > | offset: 51 CreateTime: 1593570958044 keysize: 3 valuesize: 30 sequence:
    > -1 headerKeys: [] key: ... payloadSchemaId: 2 payload:
    > .iRKoMVeoVVnTmQEuqwSTHZQ
    > baseOffset: 52 lastOffset: 52 count: 1 baseSequence: -1 lastSequence: -1
    > producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional:
    > false isControl: false position: 3019 CreateTime: 1593570969001 size: 84
    > magic: 2 compresscodec: NONE crc: 2188466405 isvalid: true
    > ```
    >
    > -Mohan
    >
Mime
View raw message