LoggingMessageFormatter с kafka-avro-console-consumer

Я пытаюсь напечатать сообщения avro в теме kafka, используя kafka-avro-console-consumer в формате log4j.

Для этого я использую следующую команду kafka-avro-console-consumer:

 bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic avro-test -property print.key=true --formatter kafka.tools.LoggingMessageFormatter

Я экспортировал KAFKA_OPTS с помощью следующей команды:

 export $KAFKA_OPTS= -Dlog4j.configuration=file:/path/to/file/kafka-console-consumer-log4j.properties

Теперь, если я запустил обычный kafka-console-consumer, используя следующую команду:

bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic avro-test -property print.key=true --formatter kafka.tools.LoggingMessageFormatter

Я могу создать вывод с поддержкой log4j:

[2018-07-17 19:09:40,514] INFO [Consumer clientId=consumer-1, groupId=console-consumer-10597] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-07-17 19:09:40,522] INFO [Consumer clientId=consumer-1, groupId=console-consumer-10597] Successfully joined group with generation 1 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-07-17 19:09:40,523] INFO [Consumer clientId=consumer-1, groupId=console-consumer-10597] Setting newly assigned partitions [avro-test-0] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
[2018-07-17 19:09:40,531] INFO [Consumer clientId=consumer-1, groupId=console-consumer-10597] Resetting offset for partition avro-test-0 to offset 23. (org.apache.kafka.clients.consumer.internals.Fetcher)

Однако этот параметр форматирования не сработает, если я использую потребителя avro, используя следующую команду:

 bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic avro-test -property print.key=true --formatter kafka.tools.LoggingMessageFormatter

Он просто использует средство форматирования по умолчанию.

Может быть, мне что-то здесь не хватает?


apache-kafka avro confluent-schema-registry log4j kafka-consumer-api
person Anirudha Nanda    schedule 17.07.2018    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Я думаю, что если вы переопределите --formatter, вы больше не будете получать сообщения Avro, поскольку kafka.tools.LoggingMessageFormatter не понимает, как десериализовать Avro

Ссылка - исходный код 

DEFAULT_AVRO_FORMATTER="--formatter io.confluent.kafka.formatter.AvroMessageFormatter"

...

for OPTION in "$@"
do 
  case $OPTION in
    --formatter)
DEFAULT_AVRO_FORMATTER=""

...

exec $(dirname $0)/schema-registry-run-class kafka.tools.ConsoleConsumer $DEFAULT_AVRO_FORMATTER ...

поэтому он должен запускать kafka.tools.ConsoleConsumer --formatter kafka.tools.LoggingMessageFormatter, как и ожидалось, поскольку значение по умолчанию не назначено, а schema-registry-run-class определяет KAFKA_OPTS, но вам не нужно иметь пробелов или знаков доллара в этой строке

export KAFKA_OPTS='-Dlog4j.configuration=file:/path/to/file/kafka-console-consumer-log4j.properties'
bin/kafka-avro-console-consumer ...
person OneCricketeer    schedule 17.11.2018