From a6c14003a9dcecc2e50634e201bf184379616636 Mon Sep 17 00:00:00 2001
From: "Matthias J. Sax" <matthias@confluent.io>
Date: Thu, 26 Oct 2023 10:30:39 -0700
Subject: [PATCH] HOTFIX: close iterator to avoid resource leak (#14624)

Reviewers: Hao Li <hli@confluent.io>, Bill Bejeck <bill@confluent.io>
---
 .../KStreamSessionWindowAggregate.java        | 30 ++++++++++---------
 1 file changed, 16 insertions(+), 14 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
index 2a3f27c68fc..5f7f709a654 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
@@ -264,22 +264,24 @@ public class KStreamSessionWindowAggregate<KIn, VIn, VAgg> implements KStreamAgg
                                   final long emitRangeUpperBound) {
             final long startMs = time.milliseconds();
 
+            int emittedCount = 0;
+
             // Only time ordered (indexed) session store should have implemented
             // this function, otherwise a not-supported exception would throw
-            final KeyValueIterator<Windowed<KIn>, VAgg> windowToEmit = store
-                .findSessions(emitRangeLowerBound, emitRangeUpperBound);
-
-            int emittedCount = 0;
-            while (windowToEmit.hasNext()) {
-                emittedCount++;
-                final KeyValue<Windowed<KIn>, VAgg> kv = windowToEmit.next();
-
-                tupleForwarder.maybeForward(
-                    record.withKey(kv.key)
-                        .withValue(new Change<>(kv.value, null))
-                        // set the timestamp as the window end timestamp
-                        .withTimestamp(kv.key.window().end())
-                        .withHeaders(record.headers()));
+            try (final KeyValueIterator<Windowed<KIn>, VAgg> windowToEmit = store
+                    .findSessions(emitRangeLowerBound, emitRangeUpperBound)) {
+
+                while (windowToEmit.hasNext()) {
+                    emittedCount++;
+                    final KeyValue<Windowed<KIn>, VAgg> kv = windowToEmit.next();
+
+                    tupleForwarder.maybeForward(
+                        record.withKey(kv.key)
+                            .withValue(new Change<>(kv.value, null))
+                            // set the timestamp as the window end timestamp
+                            .withTimestamp(kv.key.window().end())
+                            .withHeaders(record.headers()));
+                }
             }
             emittedRecordsSensor.record(emittedCount);
             emitFinalLatencySensor.record(time.milliseconds() - startMs);