[Backport 2.9] Close Zstd Dictionary after execution to avoid any memory leak. (#9403)

.ci/bwcVersions

@@ -49,6 +49,7 @@ BWC_VERSION:
   - "1.3.10"
   - "1.3.11"
   - "1.3.12"
+  - "1.3.13"
   - "2.0.0"
   - "2.0.1"
   - "2.0.2"

CHANGELOG.md

@@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Removed
 
 ### Fixed
+- Fix memory leak when using Zstd Dictionary ([#9403](https://github.com/opensearch-project/OpenSearch/pull/9403))
 
 ### Security
 

libs/core/src/main/java/org/opensearch/Version.java

@@ -90,6 +90,7 @@ public class Version implements Comparable<Version>, ToXContentFragment {
     public static final Version V_1_3_10 = new Version(1031099, org.apache.lucene.util.Version.LUCENE_8_10_1);
     public static final Version V_1_3_11 = new Version(1031199, org.apache.lucene.util.Version.LUCENE_8_10_1);
     public static final Version V_1_3_12 = new Version(1031299, org.apache.lucene.util.Version.LUCENE_8_10_1);
+    public static final Version V_1_3_13 = new Version(1031399, org.apache.lucene.util.Version.LUCENE_8_10_1);
     public static final Version V_2_0_0 = new Version(2000099, org.apache.lucene.util.Version.LUCENE_9_1_0);
     public static final Version V_2_0_1 = new Version(2000199, org.apache.lucene.util.Version.LUCENE_9_1_0);
     public static final Version V_2_0_2 = new Version(2000299, org.apache.lucene.util.Version.LUCENE_9_1_0);

server/src/main/java/org/opensearch/index/codec/customcodecs/ZstdCompressionMode.java

@@ -102,11 +102,13 @@
 
                 // dictionary compression first
                 doCompress(bytes, offset, dictLength, cctx, out);
-                cctx.loadDict(new ZstdDictCompress(bytes, offset, dictLength, compressionLevel));
+                try (ZstdDictCompress dictCompress = new ZstdDictCompress(bytes, offset, dictLength, compressionLevel)) {
+                    cctx.loadDict(dictCompress);
 
-                for (int start = offset + dictLength; start < end; start += blockLength) {
-                    int l = Math.min(blockLength, end - start);
-                    doCompress(bytes, start, l, cctx, out);
+                    for (int start = offset + dictLength; start < end; start += blockLength) {
+                        int l = Math.min(blockLength, end - start);
+                        doCompress(bytes, start, l, cctx, out);
+                    }
                 }
             }
         }
@@ -169,32 +171,33 @@
 
                 // decompress dictionary first
                 doDecompress(in, dctx, bytes, dictLength);
-
-                dctx.loadDict(new ZstdDictDecompress(bytes.bytes, 0, dictLength));
-
-                int offsetInBlock = dictLength;
-                int offsetInBytesRef = offset;
-
-                // Skip unneeded blocks
-                while (offsetInBlock + blockLength < offset) {
-                    final int compressedLength = in.readVInt();
-                    in.skipBytes(compressedLength);
-                    offsetInBlock += blockLength;
-                    offsetInBytesRef -= blockLength;
+                try (ZstdDictDecompress dictDecompress = new ZstdDictDecompress(bytes.bytes, 0, dictLength)) {
+                    dctx.loadDict(dictDecompress);
+
+                    int offsetInBlock = dictLength;
+                    int offsetInBytesRef = offset;
+
+                    // Skip unneeded blocks
+                    while (offsetInBlock + blockLength < offset) {
+                        final int compressedLength = in.readVInt();
+                        in.skipBytes(compressedLength);
+                        offsetInBlock += blockLength;
+                        offsetInBytesRef -= blockLength;
+                    }
+
+                    // Read blocks that intersect with the interval we need
+                    while (offsetInBlock < offset + length) {
+                        bytes.bytes = ArrayUtil.grow(bytes.bytes, bytes.length + blockLength);
+                        int l = Math.min(blockLength, originalLength - offsetInBlock);
+                        doDecompress(in, dctx, bytes, l);
+                        offsetInBlock += blockLength;
+                    }
+
+                    bytes.offset = offsetInBytesRef;
+                    bytes.length = length;
+
+                    assert bytes.isValid() : "decompression output is corrupted";
                 }
-
-                // Read blocks that intersect with the interval we need
-                while (offsetInBlock < offset + length) {
-                    bytes.bytes = ArrayUtil.grow(bytes.bytes, bytes.length + blockLength);
-                    int l = Math.min(blockLength, originalLength - offsetInBlock);
-                    doDecompress(in, dctx, bytes, l);
-                    offsetInBlock += blockLength;
-                }
-
-                bytes.offset = offsetInBytesRef;
-                bytes.length = length;
-
-                assert bytes.isValid() : "decompression output is corrupted";
             }
         }