implement tracing streams.filter()

[jorgheymans]

fixes #899

[jeqo]

@jorgheymans thanks! I think we are very close. I left some comments to consider filterNot as well https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java#L79

[jorgheymans]

One more thing, TracingFilterTransformer is an AbstractTracingTransformer that handles its own transform logic because it needs to tag the span. Because of this, the wrapping in TracingTransformerSupplier#get() is perhaps not needed for this case ? It adds an extra span unnecessarily IMO.

[jeqo]

One more thing, TracingFilterTransformer is an AbstractTracingTransformer that handles its own transform logic because it needs to tag the span. Because of this, the wrapping in TracingTransformerSupplier#get() is perhaps not needed for this case ? It adds an extra span unnecessarily IMO.

True, we need a Supplier for TracingFilterTransformer as well.

[jorgheymans]

The last commit adds the supplier and fixes the integration tests. Also somehow the error about accessing the kafka headers outside of message processing is gone now, maybe it had something to do with the FilterTransformer being wrapped .. not sure.

So from my part this is good to go, let me know if you have any further comments i'd be happy to address them 👍

[jeqo]

LGTM, thanks @jorgheymans !!