You are viewing a plain text version of this content. The canonical link for it is here.

Posted to users@kafka.apache.org by Jorg Heymans <jh...@apache.org> on 2019/07/04 09:51:29 UTC

stream.filter() based on message header

Hi, 

I understand that it's currently not possible to access message headers from the filter() DSL operation, is there a semantically equivalent (stateless) operation that can achieve the same result ? I understand that transform(), transformValues() and process() could achieve the same result but they are stateful. 

Thanks,
Jorg

Re: stream.filter() based on message header

Posted by Jorg Heymans <jh...@apache.org>.


On 2019/07/04 12:41:58, Bruno Cadonna <br...@confluent.io> wrote: 
> Hi Jorg,
> 
> transform(), transformValues, and process() are not stateful if you do
> not add any state store to them. You only need to leave the
> variable-length argument empty.

doh i did not realize you could leave the statestore argument empty, thanks for the quick reply !

Jorg

Re: stream.filter() based on message header

Posted by Bruno Cadonna <br...@confluent.io>.

Hi Jorg,

transform(), transformValues, and process() are not stateful if you do
not add any state store to them. You only need to leave the
variable-length argument empty.

Within those methods you can implement your desired filter operation.

Best,
Bruno

On Thu, Jul 4, 2019 at 11:51 AM Jorg Heymans <jh...@apache.org> wrote:
>
> Hi,
>
> I understand that it's currently not possible to access message headers from the filter() DSL operation, is there a semantically equivalent (stateless) operation that can achieve the same result ? I understand that transform(), transformValues() and process() could achieve the same result but they are stateful.
>
> Thanks,
> Jorg
>
>

Re: stream.filter() based on message header

Posted by Sophie Blee-Goldman <so...@confluent.io>.

Hi Jorg,

The transform/transformValues/process operators are not inherently
stateful. They *can* be stateful if you choose to connect a statestore, but
otherwise they are stateless and in your case, most likely what you want to
use.

Cheers,
Sophie

On Thu, Jul 4, 2019 at 2:51 AM Jorg Heymans <jh...@apache.org> wrote:

> Hi,
>
> I understand that it's currently not possible to access message headers
> from the filter() DSL operation, is there a semantically equivalent
> (stateless) operation that can achieve the same result ? I understand that
> transform(), transformValues() and process() could achieve the same result
> but they are stateful.
>
> Thanks,
> Jorg
>
>
>