Update streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

Co-authored-by: TengYao Chi <kitingiao@gmail.com>

Author: aliehsaeedii

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

@@ -109,7 +109,8 @@ private void openWithLegacyTimestampedCF(final DBOptions dbOptions,
         // Check if legacy CF has data
         final RocksIterator legacyIter = db.newIterator(legacyCf);
         legacyIter.seekToFirst();
-        try {
+        try (final RocksIterator legacyIter = db.newIterator(legacyCf)) {
+            legacyIter.seekToFirst();
             if (legacyIter.isValid()) {
                 log.info("Opening store {} in upgrade mode", name);
                 cfAccessor = new DualColumnFamilyAccessor(legacyCf, headersCf,
@@ -119,9 +120,7 @@ private void openWithLegacyTimestampedCF(final DBOptions dbOptions,
                 cfAccessor = new SingleColumnFamilyAccessor(headersCf);
                 legacyCf.close();
             }
-        } finally {
-            legacyIter.close();
-        }
+        } 
     }
 
     /**