This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-persistence-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 83f76bfe Close the blob streams that snapshot payloads are read from
(#607)
83f76bfe is described below
commit 83f76bfe9d7b7dc0281432c6f1299bc5d526e4f8
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Sep 7 07:55:49 2026 +0100
Close the blob streams that snapshot payloads are read from (#607)
Motivation:
The Oracle snapshot column mapping converts a blob with
`blob.getBinaryStream.toArray`. `InputStreamOps.toArray` reads the stream
to the
end but never closes it, and neither does the call site, so every snapshot
that
is read from Oracle leaves a stream open. The stream is backed by driver
resources (a cursor or a temporary lob) that are only released when the
connection is returned or the garbage collector runs.
Modification:
Added `BlobOps.toArray`, which reads the blob and closes the stream in a
`finally` block, and used it in both the current and the legacy snapshot
tables.
Result:
Reading snapshots from Oracle no longer leaks a stream per row.
Tests:
- sbt "core/testOnly org.apache.pekko.persistence.jdbc.util.BlobOpsSpec" -
passed
- sbt "core/testOnly *H2*Snapshot*" - passed
- sbt "core/mimaReportBinaryIssues" - passed
- scalafmt --mode diff-ref=upstream/main - no changes
References:
None - found while reviewing the main sources for resource leaks
---
.../jdbc/snapshot/dao/SnapshotTables.scala | 4 +-
.../jdbc/snapshot/dao/legacy/SnapshotTables.scala | 4 +-
.../pekko/persistence/jdbc/util/BlobOps.scala | 41 ++++++++++++
.../pekko/persistence/jdbc/util/BlobOpsSpec.scala | 78 ++++++++++++++++++++++
4 files changed, 123 insertions(+), 4 deletions(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotTables.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotTables.scala
index ee9987e4..0d650bf7 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotTables.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotTables.scala
@@ -18,7 +18,7 @@ import org.apache.pekko
import pekko.persistence.jdbc.config.SnapshotTableConfiguration
import pekko.persistence.jdbc.snapshot.dao.SnapshotTables.SnapshotRow
import pekko.persistence.jdbc.snapshot.dao.legacy.SnapshotTables.isOracleDriver
-import pekko.persistence.jdbc.util.InputStreamOps.InputStreamImplicits
+import pekko.persistence.jdbc.util.BlobOps
object SnapshotTables {
case class SnapshotRow(
@@ -81,7 +81,7 @@ trait SnapshotTables {
import javax.sql.rowset.serial.SerialBlob
private val columnType =
- MappedColumnType.base[Array[Byte], Blob](bytes => new SerialBlob(bytes),
blob => blob.getBinaryStream.toArray)
+ MappedColumnType.base[Array[Byte], Blob](bytes => new SerialBlob(bytes),
blob => BlobOps.toArray(blob))
override val snapshotPayload: Rep[Array[Byte]] =
column[Array[Byte]](snapshotTableCfg.columnNames.snapshotPayload)(columnType)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotTables.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotTables.scala
index 847baf9f..dec59b7a 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotTables.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotTables.scala
@@ -17,7 +17,7 @@ package org.apache.pekko.persistence.jdbc.snapshot.dao.legacy
import org.apache.pekko
import pekko.persistence.jdbc.config.LegacySnapshotTableConfiguration
import pekko.persistence.jdbc.snapshot.dao.legacy.SnapshotTables.{
isOracleDriver, SnapshotRow }
-import pekko.persistence.jdbc.util.InputStreamOps._
+import pekko.persistence.jdbc.util.BlobOps
import slick.jdbc.JdbcProfile
object SnapshotTables {
@@ -62,7 +62,7 @@ trait SnapshotTables {
import javax.sql.rowset.serial.SerialBlob
private val columnType =
- MappedColumnType.base[Array[Byte], Blob](bytes => new SerialBlob(bytes),
blob => blob.getBinaryStream.toArray)
+ MappedColumnType.base[Array[Byte], Blob](bytes => new SerialBlob(bytes),
blob => BlobOps.toArray(blob))
override val snapshot: Rep[Array[Byte]] =
column[Array[Byte]](snapshotTableCfg.columnNames.snapshot)(columnType)
}
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/util/BlobOps.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/util/BlobOps.scala
new file mode 100644
index 00000000..caed5757
--- /dev/null
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/util/BlobOps.scala
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.util
+
+import java.sql.Blob
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.persistence.jdbc.util.InputStreamOps.InputStreamImplicits
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[jdbc] object BlobOps {
+
+ /**
+ * Reads the complete content of the blob and closes the stream that was
used to read it. A stream that is left
+ * open keeps driver resources (a cursor or a temporary lob) allocated for
every row that is read.
+ */
+ def toArray(blob: Blob): Array[Byte] = {
+ val inputStream = blob.getBinaryStream
+ try inputStream.toArray
+ finally inputStream.close()
+ }
+}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/util/BlobOpsSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/util/BlobOpsSpec.scala
new file mode 100644
index 00000000..c23736b4
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/util/BlobOpsSpec.scala
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.util
+
+import java.io.{ ByteArrayInputStream, InputStream }
+
+import javax.sql.rowset.serial.SerialBlob
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object BlobOpsSpec {
+
+ /**
+ * A blob that keeps track of the binary streams it handed out, like the
ones a JDBC driver creates for a lob that
+ * is read from the database.
+ */
+ class TrackingBlob(bytes: Array[Byte]) extends SerialBlob(bytes) {
+ @volatile var openStreams: Int = 0
+
+ override def getBinaryStream: InputStream = {
+ openStreams += 1
+ new ByteArrayInputStream(bytes) {
+ override def close(): Unit = {
+ openStreams -= 1
+ super.close()
+ }
+ }
+ }
+ }
+}
+
+class BlobOpsSpec extends AnyWordSpecLike with Matchers {
+ import BlobOpsSpec._
+
+ "BlobOps.toArray" must {
+ "read the content of the blob and close the stream it read from" in {
+ val bytes = Array[Byte](1, 2, 3, 4, 5)
+ val blob = new TrackingBlob(bytes)
+
+ BlobOps.toArray(blob) shouldBe bytes
+
+ blob.openStreams shouldBe 0
+ }
+
+ "close the stream when reading fails" in {
+ val blob = new TrackingBlob(Array[Byte](1, 2, 3)) {
+ override def getBinaryStream: InputStream = {
+ val delegate = super.getBinaryStream
+ new InputStream {
+ override def read(): Int = throw new RuntimeException("boom")
+ override def read(b: Array[Byte], off: Int, len: Int): Int = throw
new RuntimeException("boom")
+ override def close(): Unit = delegate.close()
+ }
+ }
+ }
+
+ a[RuntimeException] should be thrownBy BlobOps.toArray(blob)
+
+ blob.openStreams shouldBe 0
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
