Can i do some stateful operations in peek or filter or branch of kafka stream apps?

The operations peek(), filter(), and branch() are inherently stateless. When you say:

I wanna do some query, and filter messages base the results

than it depends what you want to query? It's possible (but not recommended) to query an "external" API. However, there is no built-in support for it, and there are many corner case to consider to make it robust. Note thought, that querying an external system does not make the operation stateful.

If you want to work with state, you can use transform() (and siblings) and build custom operators. If you name all your downstream operators (via Named and similar) you can use context.forward(..., To.child(...)) to implement a custom branch. For filtering you can return null to not forward anything.

Not sure what a stateful peek() would be used for, but you could also do that.

Depending on the use-case, it's also possible to implement a "stateful filter" via a stream-table join or stream-globalTable join.

IMO, the best ways to do this is using table lookup using KStream#...join or using Processor API to get access to the underlying state store (using a KStream#transformValues).

You could do that but the code will be very nasty (would not recommend this), but you can only get a read only access to a ReadOnlyKeyValueStore after Stream state had move from REBALANCING to RUNNING:

kafkaStreams.setStateListener((newState, oldState) -> {
    if (newState == KafkaStreams.State.RUNNING && oldState == KafkaStreams.State.REBALANCING) {
        ReadOnlyKeyValueStore<Object, Object> kvStore = kafkaStreams.store("stateStore", QueryableStoreTypes.keyValueStore());
        //assign this kvStore to some place so you can later using this referrer access this in filter or in peek
    }
});