Kafka Streams runner: bundles are not bounded by time

`--maxBundleTimeMs` is accepted but has no effect. A bundle is closed when it reaches `--maxBundleSize` elements or when a watermark arrives, so on a sparse stream the elements already fed to a bundle are not emitted until the next watermark, and the watermark cannot advance past them.

The natural implementation — closing the bundle from a wall-clock punctuator — produces duplicate output against a real broker. With it enabled, an integration test running two chained GroupByKeys across four partitions emits its single group six times, reproducibly, and the count keeps climbing after the input has stopped.

What has been ruled out:

- **Metrics folding.** Splitting the same input across many bundles does not change a user counter (`MetricsAcrossBundlesTest`).
- **`ProcessorContext.commit()`.** The duplication happens with the commit request removed, and does not happen with the punctuator disabled but the commit still requested.

So it appears to be closing a Fn-API bundle from a punctuator rather than from record processing.

~~Note that Kafka Streams sets `commitOffsetNeeded` inside `StreamTask#process()`, so data produced from a punctuator sits outside that accounting; KAFKA-6906 was a bug of that shape.~~

**That note was wrong — see the comment below.** There is no `commitOffsetNeeded` flag; it is `commitNeeded`, and it is set after a punctuation runs as well as from `process()`, so punctuator output is not outside the commit accounting. Corrected by Matthias J. Sax on the Kafka dev list. The cause of the duplication is still unknown, and the way forward is data-driven bundle markers rather than a wall-clock punctuator.

Part of #18479.

Start with MetricsAcrossBundlesTest and the Kafka Streams runner path that closes Fn-API bundles from record processing or punctuators. Review StreamTask#process() and the corrected commit accounting, then reproduce the two-chained-GroupByKeys case across four partitions. Done means maxBundleTimeMs bounds sparse-stream bundles without duplicate output, using data-driven bundle markers rather than a wall-clock punctuator.

Written by the indexing model from the issue text.