[NSE-400] Fix memory leak for native C2R and R2C.

native-sql-engine/core/src/main/scala/com/intel/oap/expression/ConverterUtils.scala

@@ -504,14 +504,26 @@ object ConverterUtils extends Logging {
   @throws[IOException]
   def getSchemaBytesBuf(schema: Schema): Array[Byte] = {
     val out: ByteArrayOutputStream = new ByteArrayOutputStream
-    MessageSerializer.serialize(new WriteChannel(Channels.newChannel(out)), schema)
-    out.toByteArray
+    var schemaBytes: Array[Byte] = null
+    try {
+      MessageSerializer.serialize(new WriteChannel(Channels.newChannel(out)), schema)
+      schemaBytes = out.toByteArray
+    } finally {
+      out.close()
+    }
+    schemaBytes
   }
 
   @throws[IOException]
   def getSchemaFromBytesBuf(schema: Array[Byte]): Schema = {
     val in: ByteArrayInputStream = new ByteArrayInputStream(schema)
-    MessageSerializer.deserializeSchema(new ReadChannel(Channels.newChannel(in)))
+    var result: Schema = null
+    try {
+      result = MessageSerializer.deserializeSchema(new ReadChannel(Channels.newChannel(in)))
+    } finally {
+      in.close()
+    }
+    result
   }
 
   @throws[GandivaException]

native-sql-engine/cpp/src/jni/jni_wrapper.cc

@@ -1511,6 +1511,7 @@ Java_com_intel_oap_vectorized_ArrowRowToColumnarJniWrapper_nativeConvertRowToCol
 
   jbyteArray serialized_record_batch =
       JniGetOrThrow(ToBytes(env, rb), "Error deserializing message");
+  env->ReleaseLongArrayElements(row_length, in_row_length, JNI_ABORT);
   return serialized_record_batch;
 }