add Header support to filter?

[jiming]

Hi,

Kafka already support headers for record. Which is a very useful feature to support filter message?

What about add the support in the RecordFilter ?

Thanks!
Jiming

[Crim]

Hey Jiming!

So you're suggesting adjusting the interface to look something like:

    /**
     * Define the filter behavior.
     * A return value of TRUE means the record WILL be shown.
     * A return value of FALSE means the record will NOT be shown.
     *
     * @param topic Name of topic the message came from.
     * @param partition Partition the message came from.
     * @param offset Offset the message came from.
     * @param key Deserialized Key object.
     * @param value Deserialized Value object.
     * @param headers Headers associated with the record.
     * @return True means the record WILL be shown.  False means the record will NOT be shown.
     */
    boolean includeRecord(final String topic, final int partition, final long offset, final Object key, final Object value, final Map<String, byte[]> headers);

?

[Crim]

Actually it looks like there's no constraint on Kafka's current implementation of Headers to ensure that header keys are unique, so perhaps using a Map here is not the right answer.

[jiming]

@Crim ,

I think you can add a method without the key value, and this method is exeucted before includeRecord. In that case we can save the cpu to deserialize the key and value.

Usually the header contains much smaller values. In that case

boolean includeRecordByHeader(final String topic, final int partition, final long offset, final Map<String, byte[]> headers) {
    // you can use Headers instead of Map here
     if(headers.contains('isMan') && headers.get('isMan').equals("true")){
        return true;
    } else {
        return false;
    }

}

[Crim]

So the RecordFilter is invoked after the key & message have already been deserialized by the underlying kafka consumer, it's injected using Kafka's interceptor interface.

https://github.com/SourceLabOrg/kafka-webview/blob/master/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/filter/RecordFilterInterceptor.java#L53

[jiming]

I see.

Then perhaps using your proposal is fine. Only change the Map to Headers. IMO used for filter is one major purpose of headers.

boolean includeRecord(final String topic, final int partition, final long offset, final Object key, final Object value, final Headers headers);