是否可以使用Kafka Stream来统计不同的事件(按id)？

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Serialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;

import java.util.Properties;

public class EventCountingApp {
    public static void main(String[] args) {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "event-counting-app");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> events = builder.stream("events-topic");
        KGroupedStream<String, String> groupedEvents = events.groupByKey();

        KTable<Windowed<String>, Long> eventCounts = groupedEvents.windowedBy(TimeWindows.of(5000))
                .count(Materialized.as("event-counts"));

        eventCounts.toStream().foreach((windowedId, count) -> {
            String id = windowedId.key();
            long windowStart = windowedId.window().start();
            long windowEnd = windowedId.window().end();
            System.out.println("Event count for id " + id + " in window [" + windowStart + ", " + windowEnd + "] is " + count);
        });

        KafkaStreams streams = new KafkaStreams(builder.build(), config);
        streams.start();
    }
}