KTools: Console Consumer Filter

First written March 2024, last updated September 2026.

A second command-line tool has been added to KTools. The command kafka-console-consumer-filter allows for filtering JSON, JSON Schema, and Avro serialized messages by JSON Path expressions. In addition to filtering a message, JSON Paths can also be used for highlighting elements within the JSON.

Introduction

This tool is designed around questions like:

• Did an event with a given JSON Element make it to Kafka?
• How do we quickly check events on the topic over the past 2 hours without having to write an ISO 8601 string?
• How do we stream data as it is happening, highlighting what we’re most interested in?

Why

As developers, we embrace command-line tools. The ability to see and display content on a topic is important for troubleshooting, validation, and onboarding new developers. In the past, we would chain kafka-console-consumer with Unix tools like grep, but that had its struggles.

We also found date-time navigation with the standard CLI tools frustrating. Not being able to stop consumption at a given time also made it quite challenging to verify a simple question, “Was a message created two to three hours ago?”

Overview

The best way to give an overview of this tool is through examples. Here are a few to see if this CLI tool can help you.

Find messages with a price under 1.00

Specifically

• Filter any message that doesn’t have a price element that is less than 1.00.
• Highlight the items in blue, the SKU in yellow, prices less than 1.00 in red, and any price greater than 30.00 in green.
• Search the entire topic, exiting when it reaches offsets associated with “right now”.
• Include the topic key and metadata as part of the output.

Syntax

kafka-console-consumer-filter --bootstrap-server localhost:9092 --topic orders-pickup \    --filter "$..[?(@.price < 1.0)]" \    --highlight "BLUE=$..items" \    --highlight "RED=$..[?(@.price < 1.0)].price" \    --highlight "GREEN=$..[?(@.price > 30.0)].price" \    --highlight "YELLOW=$..sku" \    --include-key \    --include-metadata \    --start earliest \    --end now

Output

Find messages with 4 or more line items

Specifically

• Filter only orders that have at least 4 elements in the “items” array.
• Highlight the item array as blue, prices with the default color (red), and all name attributes in purple.
• Start from 1 day ago and stop when messages exist after the time this command starts executing.

Syntax

kafka-console-consumer-filter --bootstrap-server localhost:9092 --topic orders-pickup \    --filter "$..[?( @.items.length() >= 4 )]" \    --highlight "BLUE=$..items" \    --highlight "$..price" \    --highlight "PURPLE=$..name" \    --rewind 1d \    --end now

Output

Display Avro as JSON with orderIds and prices highlighted

Specifically

• Select avro as the format, providing the URL to the schema registry.
• Start from March 11th, 2024 at 17:00 UTC, and stop at the time when the command started.
• Highlight orderIds in blue and all price elements in red.

Syntax

kafka-console-consumer-filter --bootstrap-server localhost:9092 --topic purchase-orders \    --format avro \    --schema-registry http://localhost:8081 \    --highlight "BLUE=$..orderId" \    --highlight "$..price" \    --start "2024-03-11T17Z" \    --end now

Output

The flags you will use most

• --filter is a JSON Path expression that must match the message value for the message to be displayed. For Avro, it runs against the JSON rendering of the record.
• --highlight (repeatable) is COLOR=Path; omit the color to use the default (red). Colors: black, red, green, yellow, blue, purple, cyan, and white.
• --start takes a flexible date-time (yyyy-MM-dd(T, )HH:mm:ss.SSSZ) or now, earliest, or latest; --rewind 2h moves the start time back by a duration. Timezone defaults to the system timezone, so use Z if you mean UTC.
• --end and --forward set the end time the same way; --end now stops when the consumer reaches offsets from when the command started.
• --format is json, json-schema, or avro; the Schema Registry formats take --schema-registry.
• --include-key and --include-metadata add the key and the topic, partition, offset, and timestamp to the output JSON, so JSON Path can filter on them. The value is then nested under value, so adjust your JSON Path expressions.

Connection options mirror the standard Kafka CLI tools; both --consumer.config and --consumer-config are accepted, because we always forget whether the middle delimiter is . or -. The tool assigns partitions directly rather than subscribing, so it never commits consumer offsets. The full option reference is in the README.

Project

This CLI tool is the second tool to be added to the KTools open-source project. For more details on this project, please see the release page for the first tool kafka-topic-truncate.