feat: deterministic close on the wrappers that own native state

zenoh-java/src/commonMain/kotlin/io/zenoh/Config.kt

@@ -34,13 +34,14 @@ import java.nio.file.Path
  *
  * A default configuration can be loaded using [Config.loadDefault].
  *
- * The native configuration a Config wraps is consumed by [Zenoh.open] (or [Zenoh.scout]); a Config that is
- * never used is released by the garbage-collection backstop.
+ * [Zenoh.open] (and [Zenoh.scout]) copy the native configuration rather than taking it, so a Config stays
+ * usable afterwards and its native memory is released by [close] — or, if never closed, by the
+ * garbage-collection backstop.
  *
  * Visit the [default configuration](https://github.com/eclipse-zenoh/zenoh/blob/main/DEFAULT_CONFIG.json5) for more
  * information on the Zenoh config parameters.
  */
-class Config internal constructor(internal val zConfig: JniConfig) {
+class Config internal constructor(internal val zConfig: JniConfig) : AutoCloseable {
 
     companion object {
 
@@ -145,4 +146,11 @@ class Config internal constructor(internal val zConfig: JniConfig) {
     @Throws(ZError::class)
     fun insertJson5(key: String, value: String) =
         zConfig.insertJson5(key, value, throwZError0, throwZError)
+
+    /**
+     * Releases the native configuration. Idempotent. Any later use of this
+     * Config — [getJson], [insertJson5], or opening a session with it — fails
+     * with a [ZError] reporting a closed handle.
+     */
+    override fun close() = zConfig.close()
 }

zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/ZBytes.kt

@@ -14,6 +14,7 @@
 
 package io.zenoh.bytes
 
+import io.zenoh.exceptions.ZError
 import io.zenoh.exceptions.throwZError0
 import io.zenoh.jni.bytes.ZBytes as JniZBytes
 
@@ -44,13 +45,17 @@ import io.zenoh.jni.bytes.ZBytes as JniZBytes
  * garbage-collection backstop — payloads are the per-message hot path,
  * and registering a GC cleaner per message measured −23% throughput at
  * small payload sizes. In callback-based subscribers/queryables, access
- * (or discard) payloads and attachments you care about; unread ones on
+ * (or [discard]) payloads and attachments you care about; unread ones on
  * dropped samples are the one place native memory can be retained.
+ *
+ * A payload you are not going to read can be released explicitly with
+ * [discard] / [close] (try-with-resources in Java, `use` in Kotlin) — see
+ * [discard] for what that means in each state.
  */
 class ZBytes private constructor(
     initialBytes: ByteArray?,
     private var handle: JniZBytes?,
-) : IntoZBytes {
+) : IntoZBytes, AutoCloseable {
 
     /**
      * The materialized bytes, `null` until a handle-backed ZBytes is read.
@@ -68,11 +73,13 @@ class ZBytes private constructor(
      * LAZILY on first access — one borrow-copy out of the native buffer via
      * `zZbytesToBytes` — then closes the native handle (forward-extraction
      * rule: the handle is delivered eagerly, the heavy bytes on demand).
+     *
+     * Throws [ZError] if this ZBytes was [discard]ed before being read.
      */
     internal val bytes: ByteArray
         get() = eager ?: synchronized(this) {
             eager ?: run {
-                val h = handle!!
+                val h = handle ?: throw ZError("ZBytes was discarded before its bytes were read.")
                 val b = h.toBytes(throwZError0)
                 eager = b
                 handle = null
@@ -81,6 +88,26 @@ class ZBytes private constructor(
             }
         }
 
+    /**
+     * Releases the native buffer of an **unread** received payload without
+     * copying it out. Idempotent, and safe against a concurrent read: it takes
+     * the same monitor as the lazy materialization, so either the read wins
+     * (the bytes are materialized and stay readable) or the discard wins (any
+     * later read fails with [ZError]).
+     *
+     * On a ZBytes whose bytes are already available — one built by [from], or a
+     * received one already read — there is no native memory left to release and
+     * this does nothing: the bytes stay readable.
+     */
+    @Synchronized
+    fun discard() {
+        handle?.close()
+        handle = null
+    }
+
+    /** Equivalent to [discard]; lets a payload be used with `use` / try-with-resources. */
+    override fun close() = discard()
+
     companion object {
 
         /**

zenoh-java/src/commonMain/kotlin/io/zenoh/session/SessionDeclaration.kt

@@ -20,8 +20,15 @@ package io.zenoh.session
  * A session declaration is either a [io.zenoh.pubsub.Publisher],
  * a [io.zenoh.pubsub.Subscriber] or a [io.zenoh.query.Queryable] declared from a [io.zenoh.Session].
  */
-interface SessionDeclaration {
+interface SessionDeclaration : AutoCloseable {
 
     /** Undeclare a declaration. No further operations should be performed after calling this function. */
     fun undeclare()
+
+    /**
+     * Equivalent to [undeclare], so a declaration held through this interface can be used with
+     * try-with-resources (Java) or `use` (Kotlin). Every implementation in this SDK overrides it
+     * with the same behaviour.
+     */
+    override fun close() = undeclare()
 }

zenoh-java/src/jvmTest/java/io/zenoh/ConfigTest.java

@@ -349,4 +349,14 @@ public void insertIllFormattedJson5ShouldFailTest() throws ZError {
 
         assertTrue(retrievedEndpoints.contains("8.8.8.8"));
     }
+
+    @Test
+    public void closeIsIdempotentAndInvalidatesTheConfigTest() {
+        Config config = Config.loadDefault();
+        config.close();
+        config.close();
+
+        assertThrows(ZError.class, () -> config.getJson("mode"));
+        assertThrows(ZError.class, () -> config.insertJson5("mode", "\"peer\""));
+    }
 }

zenoh-java/src/jvmTest/kotlin/io/zenoh/ZBytesLifecycleTest.kt

@@ -0,0 +1,120 @@
+//
+// Copyright (c) 2026 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
+//
+
+package io.zenoh
+
+import io.zenoh.bytes.ZBytes
+import io.zenoh.exceptions.ZError
+import io.zenoh.exceptions.throwZError0
+import io.zenoh.session.SessionDeclaration
+import java.util.concurrent.CyclicBarrier
+import io.zenoh.jni.bytes.ZBytes as JniZBytes
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertThrows
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+/**
+ * The explicit lifecycle of a received payload: it owns a native buffer until
+ * its bytes are materialized (or discarded), and — unlike every other
+ * handle-owning class here — has no garbage-collection backstop, so
+ * [ZBytes.discard]/[ZBytes.close] is the only way to release one that is never
+ * read. A handle-backed instance is built here directly rather than through a
+ * session: what is under test is the wrapper's state machine, not delivery.
+ */
+class ZBytesLifecycleTest {
+
+    private val payload = "the payload".encodeToByteArray()
+
+    private fun received(): ZBytes = ZBytes.fromHandle(JniZBytes.newFromVec(payload, throwZError0))
+
+    @Test
+    fun discardBeforeMaterializationReleasesTheBufferAndInvalidatesTheBytes() {
+        val zbytes = received()
+        zbytes.discard()
+
+        assertThrows(ZError::class.java) { zbytes.toBytes() }
+    }
+
+    @Test
+    fun discardAfterMaterializationKeepsTheBytesReadable() {
+        val zbytes = received()
+        assertArrayEquals(payload, zbytes.toBytes())
+
+        zbytes.discard()
+
+        assertArrayEquals(payload, zbytes.toBytes())
+    }
+
+    @Test
+    fun discardAndCloseAreIdempotent() {
+        val unread = received()
+        unread.close()
+        unread.close()
+        unread.discard()
+
+        val read = received()
+        read.toBytes()
+        read.close()
+        read.discard()
+    }
+
+    @Test
+    fun closingAValueZBytesIsANoOp() {
+        val zbytes = ZBytes.from(payload)
+        zbytes.close()
+
+        assertArrayEquals(payload, zbytes.toBytes())
+    }
+
+    @Test
+    fun concurrentReadAndCloseHaveDeterministicOutcomes() {
+        repeat(500) {
+            val zbytes = received()
+            val barrier = CyclicBarrier(2)
+            var failure: Throwable? = null
+
+            val reader = Thread {
+                barrier.await()
+                try {
+                    assertArrayEquals(payload, zbytes.toBytes())
+                } catch (e: ZError) {
+                    // The discard won the race: a documented outcome.
+                } catch (e: Throwable) {
+                    failure = e
+                }
+            }
+            val closer = Thread {
+                barrier.await()
+                try {
+                    zbytes.close()
+                } catch (e: Throwable) {
+                    failure = e
+                }
+            }
+
+            reader.start()
+            closer.start()
+            reader.join()
+            closer.join()
+
+            failure?.let { throw it }
+        }
+    }
+
+    @Test
+    fun aSessionDeclarationIsAutoCloseable() {
+        assertTrue(AutoCloseable::class.java.isAssignableFrom(SessionDeclaration::class.java))
+    }
+}