说明：我只想处理密钥的最新唯一事件。我有kafkastreams<string，locationchangeevent>kstreams。假设我在Kafka团队里有这些事件：

{id= "DELHI", event1},
{id= "MUMBAI", event2},
{id= "DELHI", event3},
{id= "JAIPUR", event4},
{id= "MUMBAI", event5}

现在，我想对它们进行分组（比如10分钟内），这样我就只能在给定的时间范围内为每个关键点生成最新的事件。

`**EXPECTED EVENTS:**{id= "DELHI", event3},
                     {id= "MUMBAI", event5},
                     {id= "JAIPUR", event4}`

**Events output according to my code implementation :**

            {id= "DELHI", event1},
            {id= "MUMBAI", event2},
            {id= "JAIPUR", event4} and rest are marked as duplicated.

根据下面所附的代码，我可以将第一个唯一的事件推送给消费者，并在给定的时间内用相同的键将所有事件进一步标记为重复。但我不想发送第一个事件，而是要发送每个键的最新事件，即该时间范围内每个键的最后一个事件。

Properties streamsConfiguration = this.buildKafkaProperties();
                         StreamsBuilder builder = new StreamsBuilder(); KStream<String, LocationChangeEvent> kStream = builder.stream(this.kafkaConfigProperties.getTopicName(), Consumed.with(Serdes.String(), locationChangeEventSerde));
                            final StoreBuilder<WindowStore<String, LocationChangeEvent>> dedupStoreBuilder = Stores.windowStoreBuilder(
                                        Stores.persistentWindowStore(storeName, retentionPeriod, numSegment, minutes, false),
                                        Serdes.String(),
                                        serde);
                                    builder.addStateStore(dedupStoreBuilder);
                                    kStream.filter((key,value)->value.getChangeId()!=null)
                                        .transformValues(() -> new DeduplicationTransformer<>(windowSize.toMillis(), (key, value) -> key), storeName)
                                        .filter((k, v) -> v != null)
                                        .foreach((requestId, object) -> {
                                                    // below function push event to consumer
                                                     this.streamKafkaFunction(object);
                                        });
                            KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
                                    streams.start();
                    }
                 private static class DeduplicationTransformer<K, V, E> implements ValueTransformerWithKey<K, V, V> {
                            private ProcessorContext context;
                            private WindowStore<E, Long> eventIdStore;
                            private final long leftDurationMs;
                            private final long rightDurationMs;
                            private final KeyValueMapper<K, V, E> idExtractor;
                    DeduplicationTransformer(final long maintainDurationPerEventInMs, final KeyValueMapper<K, V, E> idExtractor) {
                                if (maintainDurationPerEventInMs < 1) {
                                    throw new IllegalArgumentException("maintain duration per event must be >= 1");
                                }
                                leftDurationMs = maintainDurationPerEventInMs;
                                rightDurationMs = maintainDurationPerEventInMs;
                                this.idExtractor = idExtractor;
                            }

                            @SuppressWarnings("unchecked")
                            @Override
                            public void init(ProcessorContext context) {
                                this.context = context;
                                eventIdStore = (WindowStore<E, Long>) 
                                context.getStateStore(storeName);
                            }
                        @Override
                            public V transform(final K key, final V value) {
                                final E eventId = idExtractor.apply(key, value);
                                LOGGER.info("Event is : {}", eventId);
                                if (eventId == null) {
                                    return value;
                                } else {
                                    final V output;
                                    if (isDuplicate(eventId)) {
                                        output = null;
                                    } else {
                                        output = value;
                                        rememberNewEvent(eventId, context.timestamp());
                                    }
                                    return output;
                                }
                            }
                    private boolean isDuplicate(final E eventId) {
                                final long eventTime = context.timestamp();
                                final WindowStoreIterator<Long> timeIterator = eventIdStore.fetch(
                                    eventId,
                                    eventTime - leftDurationMs,
                                    eventTime + rightDurationMs);
                                final boolean isDuplicate = timeIterator.hasNext();
                                timeIterator.close();
                                return isDuplicate;
                            }

                            private void rememberNewEvent(final E eventId, final long timestamp) {
                                eventIdStore.put(eventId, timestamp, timestamp);
                            }

                    @Override
                            public void close() {
                            }
                        }