io.confluent.kafka.formatter.protobuf

Class ProtobufMessageFormatter

java.lang.Object

io.confluent.kafka.formatter.SchemaMessageFormatter<com.google.protobuf.Message>

io.confluent.kafka.formatter.protobuf.ProtobufMessageFormatter

All Implemented Interfaces:

Closeable, AutoCloseable, org.apache.kafka.common.Configurable, org.apache.kafka.common.MessageFormatter

public class ProtobufMessageFormatter
extends SchemaMessageFormatter<com.google.protobuf.Message>

Example To use ProtobufMessageFormatter, first make sure that Zookeeper, Kafka and schema registry server are all started. Second, make sure the jar for ProtobufMessageFormatter and its dependencies are included in the classpath of kafka-console-consumer.sh. Then run the following command.

1. To read only the value of the messages in JSON bin/kafka-console-consumer.sh --consumer.config config/consumer.properties --topic t1 \ --bootstrap-server localhost:9092 --formatter io.confluent.kafka.formatter.ProtobufMessageFormatter \ --property schema.registry.url=http://localhost:8081

2. To read both the key and the value of the messages in JSON bin/kafka-console-consumer.sh --consumer.config config/consumer.properties --topic t1 \ --bootstrap-server localhost:9092 --formatter io.confluent.kafka.formatter.ProtobufMessageFormatter \ --property schema.registry.url=http://localhost:8081 \ --property print.key=true

3. To read the key, value, and timestamp of the messages in JSON bin/kafka-console-consumer.sh --consumer.config config/consumer.properties --topic t1 \ --bootstrap-server localhost:9092 --formatter io.confluent.kafka.formatter.ProtobufMessageFormatter \ --property schema.registry.url=http://localhost:8081 \ --property print.key=true \ --property print.timestamp=true

Field Summary

Fields inherited from class io.confluent.kafka.formatter.SchemaMessageFormatter

deserializer

Constructor Summary

Constructors

Constructor and Description
ProtobufMessageFormatter() Constructor needed by kafka console consumer.

Method Summary

Modifier and Type	Method and Description
protected SchemaMessageDeserializer<com.google.protobuf.Message>	createDeserializer(org.apache.kafka.common.serialization.Deserializer keyDeserializer)
protected SchemaProvider	getProvider()
void	init(Properties props)
protected void	writeTo(String topic, Boolean isKey, org.apache.kafka.common.header.Headers headers, byte[] data, PrintStream output)

Methods inherited from class io.confluent.kafka.formatter.SchemaMessageFormatter

close, configure, writeTo

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

ProtobufMessageFormatter

public ProtobufMessageFormatter()

Constructor needed by kafka console consumer.

Method Detail

createDeserializer

protected SchemaMessageDeserializer<com.google.protobuf.Message> createDeserializer(org.apache.kafka.common.serialization.Deserializer keyDeserializer)

Specified by:

createDeserializer in class SchemaMessageFormatter<com.google.protobuf.Message>

init

public void init(Properties props)

Overrides:

init in class SchemaMessageFormatter<com.google.protobuf.Message>

writeTo

protected void writeTo(String topic,
                       Boolean isKey,
                       org.apache.kafka.common.header.Headers headers,
                       byte[] data,
                       PrintStream output)
                throws IOException

Specified by:

writeTo in class SchemaMessageFormatter<com.google.protobuf.Message>

Throws:

IOException

getProvider

protected SchemaProvider getProvider()

Specified by:

getProvider in class SchemaMessageFormatter<com.google.protobuf.Message>