Possible native memory leak. How to debug further?

Question

I implemented a new feature in a SpringBoot 3 backend. I have some strange (possibly) native memory problems and I do not know how I can best continue with debugging. I tried profiling the heap usage using visual vm which looks normal. I also do not see any OOM exceptions in my application. However, the java process was killed numerous times on an azure linux app service (2 cores, 8gb ram).

I noticed that the RSS size grows always quite a bit larger than the maximum heap size:

USER       PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
root        48 73.5 75.6 8377500 6161048 ?     SNl  18:05  47:43 java -Xms1g -Xmx3g -XX:MaxDirectMemorySize=500m -jar /home/site/wwwroot/b/app.jar --server.port=80

I experimented a lot with xms, xmx, maxdirectbuffers etc.

Just for reference my usecase:

1. Download CSVs from azure blobstorage
2. Convert to representing data class objects
3. Map to RealmObjects (kotlin-realm)
4. Save to realm
5. Zip realm and save zip
6. Clean up files (Realm.delete(config), delete lockfiles)
7. Repeat for ~ 15 times

I checked that all Input- and Outputstreams are closed.

I am not sure if it can be related to netty. I use

implementation("com.azure.spring:spring-cloud-azure-dependencies:5.1.0")
implementation("com.azure.spring:spring-cloud-azure-starter-storage-blob:5.1.0")

to stream the blobs which is done by netty. I read that netty can allocate native memory for connection pooling.

However, here are the most critical parts of the implementation:

AzureStorage:

class AzureStorageService(blobServiceClient: BlobServiceClient, val configuration: Configuration) {
    
        private val blobContainerClient = blobServiceClient.getBlobContainerClient("export")
        private val logger = LoggerFactory.getLogger(AzureStorageService::class.java)
    
        fun downloadBlobToInputStream(file: String): InputStream {
            return blobContainerClient.getBlobClient(file).openInputStream()
        }
    
        fun copyAllFromPrefix(from: String, to: String) {
            val options = ListBlobsOptions()
                .setPrefix(from)
    
            blobContainerClient.listBlobsByHierarchy("/", options, null)
                .filter { !it.isPrefix }
                .filterIsInstance<BlobItem>()
                .forEach {
                    val blobName = it.name.removePrefix("$from/")
                    val sourceBlob = blobContainerClient.getBlobClient(it.name)
                    val destinationBlob = blobContainerClient. 
                                           getBlobClient("$to/$blobName")

                   destinationBlob.copyFromUrl( 
                        "${sourceBlob.blobUrl}${configuration.azureStorageSasToken}")
                    sourceBlob.deleteIfExists()
                }
        }
    }

CsvFileReader:

class CsvFileReader(private val stream: InputStream, private val csvFormat: 
CsvFileFormat) {

    fun getLines(): List<CsvLine> {
        val reader = 
getStreamWithoutByteOrderMark(stream).bufferedReader(csvFormat.encoding)
        return reader.use { r ->
            val headerLine = r.readLine() ?: return emptyList()
            val header = CsvHeader.from(headerLine.split(csvFormat.separator))

            r.useLines { lines ->
                lines.map { line ->
                    CsvLine.from(header, line.split(csvFormat.separator))
                }.toList()
            }
        }
    }

    private fun getStreamWithoutByteOrderMark(stream: InputStream): InputStream {
        val bufferedStream = BufferedInputStream(stream)
        bufferedStream.mark(3)

        val byteOrderMark = (0xFEFF).toChar()
        val startBytes = bufferedStream.readNBytes(3)
        val possibleMark = String(startBytes, csvFormat.encoding).first()

        if (possibleMark != byteOrderMark) {
            bufferedStream.reset()
        }
        return bufferedStream
    }
}

Zipping:

fun createZip(filePath: String): InputStream {
    val realmFile = File(filePath)
    val zippedOutput = ByteArrayOutputStream()

    ZipOutputStream(BufferedOutputStream(zippedOutput)).use { zipOut ->
        FileInputStream(realmFile).use { fileIn ->
            val entry = ZipEntry(realmFile.name)
            zipOut.putNextEntry(entry)
            fileIn.copyTo(zipOut, DEFAULT_BUFFER_SIZE)
            zipOut.closeEntry()
        }
    }

    return ByteArrayInputStream(zippedOutput.toByteArray())
}

If there is anything more I can provide please let me know. I need some guidance.