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-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new b10237e6e HBase: close the scanner and connection in the source stage
(#1876)
b10237e6e is described below
commit b10237e6e1af553058e711e9c81f3c88596af73a
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 23 08:07:38 2026 +0100
HBase: close the scanner and connection in the source stage (#1876)
* HBase: close the scanner and connection in the source stage
Motivation:
HBaseSourceLogic acquires three resources and released one.
implicit val connection: Connection = connect(settings.conf)
lazy val table: Table = getOrCreateTable(...).get
override def preStart(): Unit =
val scanner = table.getScanner(scan) // local val, never closed
results = scanner.iterator()
override def postStop(): Unit =
table.close() // connection never closed
The Connection is a ConnectionImplementation: a ZooKeeper connection, an
RpcClient and a thread pool. Every materialization of HTableStage.source
leaks one for the life of the JVM. The ResultScanner is never closed
either, so the server-side scanner lease is held on the RegionServer
until it times out, on every run.
The sibling HBaseFlowStage already closes both table and connection, so
this is an omission in the source rather than a deliberate difference.
Modification:
Keep the scanner in a field and close all three in postStop, each
independently so that a failure to close one still releases the rest.
Failures are logged rather than rethrown, matching HBaseFlowStage: the
stage has already stopped, so failing here would only mask why. The
previous code called failStage from postStop, which is too late to have
any effect.
`table` is lazy, so a `tableOpened` flag records whether preStart forced
it; otherwise postStop would create a table only to close it. If preStart
fails part way — getScanner throwing on a missing table is the common
case — whatever was opened is still released.
The close sequence is extracted to HBaseSourceLogic.closeAll so it can be
tested without an HBase cluster.
Result:
Connections, tables and scanners are released when the source stops, on
both the normal and the failure path.
Tests:
Added HBaseSourceLogicSpec covering the ordinary path, a scanner that
fails to close, a table that fails to close, and all three failing. Each
asserts the later resources are still closed. Needs no HBase server,
which matters because the hbase integration tests are disabled in CI.
4 tests pass. Ran hbase/mimaReportBinaryIssues (clean) and the module
scalafmt checks.
References:
Compare HBaseFlowStage.postStop, which closes table and connection.
* HBase: apply scalafmt to HBaseSourceLogicSpec
---
.../connectors/hbase/impl/HBaseSourceStage.scala | 45 +++++++++++---
.../hbase/impl/HBaseSourceLogicSpec.scala | 71 ++++++++++++++++++++++
2 files changed, 108 insertions(+), 8 deletions(-)
diff --git
a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
index 7c8a30bcc..7861791da 100644
---
a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
+++
b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
@@ -17,7 +17,7 @@ import org.apache.pekko
import pekko.stream.{ Attributes, Outlet, SourceShape }
import pekko.stream.connectors.hbase.HTableSettings
import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler,
StageLogging }
-import org.apache.hadoop.hbase.client.{ Connection, Result, Scan, Table }
+import org.apache.hadoop.hbase.client.{ Connection, Result, ResultScanner,
Scan, Table }
import scala.util.control.NonFatal
@@ -43,26 +43,32 @@ private[hbase] final class HBaseSourceLogic[A](scan: Scan,
implicit val connection: Connection = connect(settings.conf)
lazy val table: Table = getOrCreateTable(settings.tableName,
settings.columnFamilies).get
+ // `table` is lazy, so postStop must not touch it unless preStart forced it —
+ // referencing it there would create a table only to close it.
+ private var tableOpened = false
+ private var scanner: ResultScanner = null
private var results: java.util.Iterator[Result] = null
setHandler(out, this)
override def preStart(): Unit =
try {
- val scanner = table.getScanner(scan)
+ val t = table
+ tableOpened = true
+ scanner = t.getScanner(scan)
results = scanner.iterator()
} catch {
case NonFatal(exc) =>
failStage(exc)
}
+ // Every resource is closed independently, so a failure to close one still
+ // releases the rest. Any of them may be unset if preStart failed part way.
override def postStop(): Unit =
- try {
- table.close()
- } catch {
- case NonFatal(exc) =>
- failStage(exc)
- }
+ HBaseSourceLogic.closeAll(
+ if (scanner ne null) scanner.close(),
+ if (tableOpened) table.close(),
+ connection.close())((what, exc) => log.error(exc, "Problem occurred
during {} close", what))
override def onPull(): Unit =
if (results.hasNext) {
@@ -72,3 +78,26 @@ private[hbase] final class HBaseSourceLogic[A](scan: Scan,
}
}
+
+private[impl] object HBaseSourceLogic {
+
+ /**
+ * Close the scanner, table and connection of a source stage. Each is closed
+ * independently so that a failure to close one still releases the others.
+ * Failures are reported to `onError` and not rethrown, matching the
+ * behaviour of `HBaseFlowStage`: the stage has already stopped, so failing
+ * here would only mask the reason it stopped.
+ */
+ def closeAll(closeScanner: => Unit, closeTable: => Unit, closeConnection: =>
Unit)(
+ onError: (String, Throwable) => Unit): Unit = {
+ def attempt(what: String, close: => Unit): Unit =
+ try close
+ catch {
+ case NonFatal(exc) => onError(what, exc)
+ }
+
+ attempt("scanner", closeScanner)
+ attempt("table", closeTable)
+ attempt("connection", closeConnection)
+ }
+}
diff --git
a/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceLogicSpec.scala
b/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceLogicSpec.scala
new file mode 100644
index 000000000..d4ef64ff5
--- /dev/null
+++
b/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceLogicSpec.scala
@@ -0,0 +1,71 @@
+/*
+ * 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.stream.connectors.hbase.impl
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.collection.mutable.ListBuffer
+
+class HBaseSourceLogicSpec extends AnyWordSpec with Matchers {
+
+ "HBaseSourceLogic.closeAll" should {
+
+ "close the scanner, the table and the connection" in {
+ val closed = ListBuffer.empty[String]
+ HBaseSourceLogic.closeAll(closed += "scanner", closed += "table", closed
+= "connection")((_, _) =>
+ fail("no error expected"))
+ closed.toList shouldBe List("scanner", "table", "connection")
+ }
+
+ "still close the table and the connection when the scanner fails to close"
in {
+ val closed = ListBuffer.empty[String]
+ val errors = ListBuffer.empty[String]
+ HBaseSourceLogic.closeAll(
+ throw new RuntimeException("scanner boom"),
+ closed += "table",
+ closed += "connection")((what, _) => errors += what)
+
+ // the connection is the expensive resource: it must be released even
+ // when an earlier close throws
+ closed.toList shouldBe List("table", "connection")
+ errors.toList shouldBe List("scanner")
+ }
+
+ "still close the connection when the table fails to close" in {
+ val closed = ListBuffer.empty[String]
+ val errors = ListBuffer.empty[String]
+ HBaseSourceLogic.closeAll(closed += "scanner", throw new
RuntimeException("table boom"),
+ closed += "connection")((what, _) => errors += what)
+
+ closed.toList shouldBe List("scanner", "connection")
+ errors.toList shouldBe List("table")
+ }
+
+ "report every failure and not rethrow" in {
+ val errors = ListBuffer.empty[String]
+ noException should be thrownBy {
+ HBaseSourceLogic.closeAll(
+ throw new RuntimeException("a"),
+ throw new RuntimeException("b"),
+ throw new RuntimeException("c"))((what, _) => errors += what)
+ }
+ errors.toList shouldBe List("scanner", "table", "connection")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
