kafka-python consumer

A line-oriented console consumer. Subscribes to one or more topics and prints each received record to standard output.

Examples

# Read every record on a topic until interrupted
kafka-python consumer -b localhost:9092 -t my-topic

# Join a consumer group and commit offsets automatically
kafka-python consumer -b localhost:9092 -g my-group -t my-topic

# Read from the beginning, then exit after 1s of idle
kafka-python consumer -b localhost:9092 -t my-topic \
    -C auto_offset_reset=earliest \
    -C consumer_timeout_ms=1000

# Print full ConsumerRecord (topic, partition, offset, key, value, ...)
kafka-python consumer -b localhost:9092 -t my-topic -f full

Reference

Kafka console consumer

usage: kafka-python consumer [-h] -b BOOTSTRAP_SERVERS [-S SECURITY_PROTOCOL]
                             [-M SASL_MECHANISM] [-U SASL_USER]
                             [-P SASL_PASSWORD] [-l LOG_LEVEL]
                             [-L ENABLE_LOGGER] [-D DISABLE_LOGGER]
                             [-C EXTRA_CONFIG] -t TOPICS [-g GROUP]
                             [-i GROUP_INSTANCE_ID] [-f FORMAT]
                             [--encoding ENCODING]

connection

-b, --bootstrap-servers

host:port for cluster bootstrap server. Can be provided multiple times.

-S, --security-protocol

PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL

Default: 'PLAINTEXT'

-M, --sasl-mechanism

PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512

Default: 'PLAIN'

-U, --sasl-user

-P, --sasl-password

logging

-l, --log-level

logging level, passed to logging.basicConfig

Default: 'CRITICAL'

-L, --enable-logger

enable a specific logger. Can be provided multiple times. If not provided, all loggers are enabled

-D, --disable-logger

disable a specific logger. Can be provided multiple times.

extended

-C, --extra-config

additional configuration properties for client in “key=val” format. Can be provided multiple times.

consumer options

-t, --topic

subscribe to topic

-g, --group

consumer group

-i, --group-instance-id

static group membership identifier

-f, --format

output format: str|raw|full

Default: 'str'

--encoding

encoding to use for str output decode()

Default: 'utf-8'