Kafka streams event deduplication keeping last event in window

I'm using Kafka Streams in a deduplication events problem over short time windows (<= 1 minute). First I've tried to tackle the problem by using DSL API with .suppress(Suppressed.untilWindowCloses(...)) operator but, given the fact that wall-clock time is not yet supported (I've seen the KIP 424), this operator is not viable for my use case.

Then, I've followed this official Confluent example in which low level Processor API is used and it was working fine but has one major limitation for my use-case. The single event (obtained by deduplication) is emitted at the beginning of the time window, subsequent duplicated events are "suppressed". In my use case I need the reverse of that, meaning that a single event should be emitted at the end of the window. I'm asking for suggestions on how to implement this use case with Processor API.

My idea was to use the Processor API with a custom Transformer and a Punctuator. The transformer would store in a WindowStore the distinct keys received without returning any KeyValue. Simultaneously, I'd schedule a punctuator running with an interval equal to the size of the window in the WindowStore. This punctuator will iterate over the elements in the store and forward them downstream. The following are some core parts of the logic:

DeduplicationTransformer (slightly modified from official Confluent example):

    @Override
    @SuppressWarnings("unchecked")
    public void init(final ProcessorContext context) {
        this.context = context;
        eventIdStore = (WindowStore<E, V>) context.getStateStore(this.storeName);

        // Schedule punctuator for this transformer.
        context.schedule(Duration.ofMillis(this.windowSizeMs), PunctuationType.WALL_CLOCK_TIME,
            new DeduplicationPunctuator<E, V>(eventIdStore, context, this.windowSizeMs));
    }

    @Override
    public KeyValue<K, V> transform(final K key, final V value) {
        final E eventId = idExtractor.apply(key, value);
        if (eventId == null) {
            return KeyValue.pair(key, value);
        } else {
            if (!isDuplicate(eventId)) {
                rememberNewEvent(eventId, value, context.timestamp());
            }
            return null;
        }
    }

DeduplicationPunctuator:

    public DeduplicationPunctuator(WindowStore<E, V> eventIdStore, ProcessorContext context,
        long retainPeriodMs) {
        this.eventIdStore = eventIdStore;
        this.context = context;
        this.retainPeriodMs = retainPeriodMs;
    }

    @Override
    public void punctuate(long invocationTime) {
        LOGGER.info("Punctuator invoked at {}, searching from {}", new Date(invocationTime), new Date(invocationTime-retainPeriodMs));

        KeyValueIterator<Windowed<E>, V> it =
            eventIdStore.fetchAll(invocationTime - retainPeriodMs, invocationTime + retainPeriodMs);

        while (it.hasNext()) {
            KeyValue<Windowed<E>, V> next = it.next();

            LOGGER.info("Punctuator running on {}", next.key.key());

            context.forward(next.key.key(), next.value);
            // Delete from store with tombstone
            eventIdStore.put(next.key.key(), null, invocationTime);
            context.commit();
        }

        it.close();
    }

Is this a valid approach? With the previous code, I'm running some integration tests and I've some synchronization issues. How can I be sure that the start of the window will coincide with the Punctuator's scheduled interval?

Also as an alternative approach, I was wondering (I've googled with no result), if there is any event triggered by window closing to which I can attach a callback in order to iterate over store and publish only distinct events.

Thanks.