Hi Team, We are seeing the issue often with Memory Leak:
*JDK 11* "org.apache.arrow" % "arrow-flight" % "17.0.0", "org.apache.arrow" % "arrow-vector" % "17.0.0", "org.apache.arrow" % "flight-core" % "17.0.0", 4-10-25 15:25:06.394 [main] ERROR o.apache.arrow.memory.BaseAllocator - Memory was leaked by query. Memory leaked: (10485760) Allocator(flight-client) 0/10485760/10485760/9223372036854775807 (res/actual/peak/limit) 2024-10-25 15:25:06.395 [main] ERROR o.apache.arrow.memory.BaseAllocator - Memory was leaked by query. Memory leaked: (10485760) Outstanding child allocators : Allocator(flight-client) 0/10485760/10485760/9223372036854775807 (res/actual/peak/limit) Allocator(ROOT) 0/10485760/10485760/9223372036854775807 (res/actual/peak/limit) 2024-10-25 15:25:06.396 [main] ERROR c.t.c.d.ArrowFlightDataFetcher - Failed to fetch data: Memory was leaked by query. Memory leaked: (10485760) Allocator(flight-client) 0/10485760/10485760/9223372036854775807 (res/actual/peak/limit) Exception in thread "main" com.tesco.cef.exceptions.CefFlightServerException: Failed to fetch data from Flight Server at com.tesco.cef.datafetcher.ArrowFlightDataFetcher$$anon$1.applyOrElse(ArrowFlightDataFetcher.scala:47) at com.tesco.cef.datafetcher.ArrowFlightDataFetcher$$anon$1.applyOrElse(ArrowFlightDataFetcher.scala:41) at scala.util.Failure.recover(Try.scala:233) at com.tesco.cef.datafetcher.ArrowFlightDataFetcher.fetchDataStream(ArrowFlightDataFetcher.scala:47) at com.tesco.cef.client.CefFlightDataClient.getStream(CefFlightDataClient.scala:27) at com.tesco.cef.samples.client.Main$.processStream(Main.scala:120) at com.tesco.cef.samples.client.Main$.main(Main.scala:60) at com.tesco.cef.samples.client.Main.main(Main.scala) Caused by: java.lang.IllegalStateException: Memory was leaked by query. Memory leaked: (10485760) Allocator(flight-client) 0/10485760/10485760/9223372036854775807 (res/actual/peak/limit) at org.apache.arrow.memory.BaseAllocator.close(BaseAllocator.java:501) at org.apache.arrow.flight.FlightClient.close(FlightClient.java:754) at scala.util.Using$Releasable$AutoCloseableIsReleasable$.release(Using.scala:392) at scala.util.Using$Releasable$AutoCloseableIsReleasable$.release(Using.scala:391) at scala.util.Using$.$anonfun$apply$1(Using.scala:268) at scala.util.Using$.apply(Using.scala:113) at com.tesco.cef.datafetcher.ArrowFlightDataFetcher.fetchDataStream$$anonfun$2(ArrowFlightDataFetcher.scala:40) at scala.util.Using$.$anonfun$apply$1(Using.scala:262) at scala.util.Using$.apply(Using.scala:113) at com.tesco.cef.datafetcher.ArrowFlightDataFetcher.fetchDataStream(ArrowFlightDataFetcher.scala:41) ... 4 more Suppressed: java.lang.IllegalStateException: Memory was leaked by query. Memory leaked: (10485760) def fetchDataStream(details: ObjectStoreDetails): Iterator[FlightStream] = { logger.info(s"Fetching data for S3 path: ${details.s3Path}") val ticketStr = buildTicketStr(details) logger.info(s"Generated ticket string: $ticketStr") val allocator = new RootAllocator(Long.MaxValue) val client = FlightClient.builder(allocator, Location.forGrpcInsecure(serverHost, serverPort)).build() val ticket = new Ticket(ticketStr.getBytes(StandardCharsets.UTF_8)) // Using Try for proper resource cleanup try { val stream = client.getStream(ticket) // Collect and return an iterator for each batch Iterator.continually(stream).takeWhile(_.next()) } catch { case e: FlightRuntimeException => logger.error(s"Error communicating with Flight server: ${e.getMessage}") throw new CefFlightServerException(s"Error communicating with Flight server: ${e.getMessage}", e) case e: Exception => logger.error(s"Failed to fetch data: ${e.getMessage}") throw new CefFlightServerException("Failed to fetch data from Flight Server", e) } finally { if (client != null) client.close() if (allocator != null) allocator.close() } } Sharing the above code snippet for reference Are we doing anything wrong here? how to avoid memory spillage issue? Thanks, Susmit