我怀疑，如果kafka流像其他工具一样支持固定大小的窗口。
但也有基于时间的窗口，由Kafka流支持。https://kafka.apache.org/11/documentation/streams/developer-guide/dsl-api.html#windowing
可以随时间定义窗口大小，而不是记录数。
翻滚时间窗口
滑动时间窗
会话窗口
跳跃时间窗
在您的情况下，翻滚时间窗口可以是一个选项。这些是不重叠的，固定大小的时间窗口。
例如，大小为5000ms的翻滚窗口具有可预测的窗口边界[0；5000),[5000;10000),... — 而不是[1000；6000),[6000;11000),... 或者像[1452；6452),[6452;11452),....

似乎还不存在。注意这个地方https://issues.apache.org/jira/browse/kafka-7432

你可以排队。就像下面这样，

@Component
@Slf4j
public class NormalTopic1StreamProcessor extends AbstractStreamProcessor<String> {

    public NormalTopic1StreamProcessor(KafkaStreamsConfiguration configuration) {
        super(configuration);
    }

    @Override
    Topology buildTopology() {
        KStream<String, String> kStream = streamsBuilder.stream("normalTopic", Consumed.with(Serdes.String(), Serdes.String()));
        // .peek((key, value) -> log.info("message received by stream 0"));
        kStream.process(() -> new AbstractProcessor<String, String>() {
            final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(100);
            final List<String> collection = new ArrayList<>();

            @Override
            public void init(ProcessorContext context) {
                super.init(context);
                context.schedule(Duration.of(1, ChronoUnit.MINUTES), WALL_CLOCK_TIME, timestamp -> {
                    processQueue();
                    context().commit();
                });
            }

            @Override
            public void process(String key, String value) {
                queue.add(value);
                if (queue.remainingCapacity() == 0) {
                    processQueue();
                }
            }

            public void processQueue() {
                queue.drainTo(collection);
                long count = collection.stream().peek(System.out::println).count();
                if (count > 0) {
                    System.out.println("count is " + count);
                    collection.clear();
                }
            }
        });
        kStream.to("normalTopic1");
        return streamsBuilder.build();
    }

}