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 a270eecaa Run the hbase tests against an in-JVM mini cluster (#1919)
a270eecaa is described below
commit a270eecaa0c3fa4c793d2d576eba7c01dd05c1cd
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Sep 6 11:02:35 2026 +0100
Run the hbase tests against an in-JVM mini cluster (#1919)
The hbase tests ran against harisekhon/hbase:2.1, an unmaintained image
whose newest tag is from 2018. Being stuck on a server that old is what
makes an hbase client upgrade look impossible: hbase 3 clients default
to RpcConnectionRegistry, and a 2.1 master has no ClientMetaService to
answer them.
hbase-shaded-testing-util starts a mini ZooKeeper, mini HDFS, master and
region server inside the test JVM, which is how hbase itself, Phoenix and
the Spark and Flink hbase connectors test. It runs the same code paths a
real cluster runs, and it is the flavour of testing-util that pairs with
the hbase-shaded-client we already depend on.
The client versions are unchanged at 2.6.6.
Two settings are needed to get the cluster up on a current JDK: the
master and region server web UIs are switched off, because their JAXB
stack needs java.lang opened to reflect; and the WAL provider is set to
filesystem, because the default asyncfs writer reflects into protobuf
internals in a way the shaded hbase/hadoop combination does not survive.
ZooKeeper is pinned to its default port, so a plain
HBaseConfiguration.create() reaches the cluster and the documentation
snippets stay unchanged.
Local run: 14 tests (7 scala, 7 java) pass in about 80s, no docker.
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.github/workflows/check-build-test.yml | 2 +-
docker-compose.yml | 9 ---
.../src/test/java/docs/javadsl/HBaseStageTest.java | 2 +
.../test/scala/docs/scaladsl/HBaseStageSpec.scala | 7 ++-
.../stream/connectors/hbase/HBaseMiniCluster.scala | 64 ++++++++++++++++++++++
project/Dependencies.scala | 8 +++
scripts/hbase-server.sh | 26 ---------
7 files changed, 80 insertions(+), 38 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index 7ce59122a..3dc373c60 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -120,7 +120,7 @@ jobs:
- { connector: google-cloud-storage }
- { connector: google-common }
- { connector: google-fcm }
- - { connector: hbase, pre_cmd:
'./scripts/hbase-server.sh' }
+ - { connector: hbase }
- { connector: hdfs }
- { connector: huawei-push-kit }
- { connector: influxdb, pre_cmd: 'docker
compose up -d influxdb' }
diff --git a/docker-compose.yml b/docker-compose.yml
index 11b22a28e..d1a1f5b9c 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -112,15 +112,6 @@ services:
curl -s --fail -X PUT -H "Content-Type: application/json" -d
'{"topic": "projects/pekko-connectors/topics/simpleTopic"}'
http://gcloud-pubsub-emulator:8538/v1/projects/pekko-connectors/subscriptions/simpleSubscription
curl -s --fail -X PUT
http://gcloud-pubsub-emulator:8538/v1/projects/pekko-connectors/topics/testTopic
curl -s --fail -X PUT -H "Content-Type: application/json" -d
'{"topic": "projects/pekko-connectors/topics/testTopic"}'
http://gcloud-pubsub-emulator:8538/v1/projects/pekko-connectors/subscriptions/testSubscription
- hbase:
- image: harisekhon/hbase:2.1
- hostname: hbase
- ports:
- - 2181:2181
- - 16000:16000
- - 16010:16010
- - 16020:16020
- - 16030:16030
ibmmq:
image: icr.io/ibm-messaging/mq:9.4.5.1-r1
environment:
diff --git a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
index 0e23b7bae..b75314d85 100644
--- a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
+++ b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
@@ -31,6 +31,7 @@ import org.apache.pekko.Done;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.japi.Pair;
+import org.apache.pekko.stream.connectors.hbase.HBaseMiniCluster;
import org.apache.pekko.stream.connectors.hbase.HTableSettings;
import org.apache.pekko.stream.connectors.hbase.javadsl.HTableStage;
import
org.apache.pekko.stream.connectors.testkit.javadsl.LogCapturingExtension;
@@ -51,6 +52,7 @@ public class HBaseStageTest {
@BeforeAll
public static void setup() {
+ HBaseMiniCluster.start();
system = ActorSystem.create();
}
diff --git a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
index 7459f88bc..8ae2b7c07 100644
--- a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
+++ b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
@@ -16,7 +16,7 @@ package docs.scaladsl
import org.apache.pekko
import pekko.Done
import pekko.actor.ActorSystem
-import pekko.stream.connectors.hbase.HTableSettings
+import pekko.stream.connectors.hbase.{ HBaseMiniCluster, HTableSettings }
import pekko.stream.connectors.hbase.scaladsl.HTableStage
import pekko.stream.connectors.testkit.scaladsl.LogCapturing
import pekko.stream.scaladsl.{ Sink, Source }
@@ -150,7 +150,7 @@ class HBaseStageSpec
"append to a cell through a flow" in {
val appendSettings = tableSettings.withConverter(appendHBaseConverter)
- // unique row per run: the sbt cross-build reruns this suite against the
same HBase instance
+ // unique row per run, so that the assertions below do not depend on
suite ordering
val id = randomRowId()
val f = Source(List(Person(id, "-a"), Person(id, "-b")))
.via(HTableStage.flow(appendSettings))
@@ -215,6 +215,9 @@ class HBaseStageSpec
private def readRow(id: Int) =
HTableStage.source(new Scan(new Get(Bytes.toBytes(s"id_$id"))),
tableSettings).runWith(Sink.seq)
+ override def beforeAll(): Unit =
+ HBaseMiniCluster.start()
+
override def afterAll(): Unit =
TestKit.shutdownActorSystem(system)
}
diff --git
a/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/HBaseMiniCluster.scala
b/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/HBaseMiniCluster.scala
new file mode 100644
index 000000000..0a80cb670
--- /dev/null
+++
b/hbase/src/test/scala/org/apache/pekko/stream/connectors/hbase/HBaseMiniCluster.scala
@@ -0,0 +1,64 @@
+/*
+ * 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
+
+import org.apache.hadoop.hbase.{ HBaseConfiguration, HBaseTestingUtility,
HConstants }
+
+/**
+ * An HBase cluster (mini ZooKeeper, mini HDFS, master and region server)
running inside the test JVM,
+ * started once and shared by every suite in that JVM.
+ *
+ * ZooKeeper is pinned to its default port so that a plain
`HBaseConfiguration.create()`, which is what
+ * the documentation snippets show, reaches the cluster without any
test-specific configuration.
+ */
+object HBaseMiniCluster {
+
+ private lazy val cluster: HBaseTestingUtility = {
+ val conf = HBaseConfiguration.create()
+ // HBaseTestingUtility reads the port it should hand to the mini ZooKeeper
from this key
+ conf.setInt("test.hbase.zookeeper.property.clientPort",
HConstants.DEFAULT_ZOOKEEPER_CLIENT_PORT)
+ // the master and region server web UIs are of no use here, and their JAXB
stack cannot
+ // initialize on a JDK that keeps java.lang closed; -1 switches them off
+ conf.setInt(HConstants.MASTER_INFO_PORT, -1)
+ conf.setInt(HConstants.REGIONSERVER_INFO_PORT, -1)
+ // the default asyncfs WAL reflects into protobuf internals in a way that
does not survive the
+ // shaded hbase/hadoop combination we test against; the filesystem WAL is
equivalent for our purposes
+ conf.set("hbase.wal.provider", "filesystem")
+ conf.set("hbase.wal.meta_provider", "filesystem")
+ val util = new HBaseTestingUtility(conf)
+ util.startMiniCluster()
+ val port = util.getZkCluster.getClientPort
+ if (port != HConstants.DEFAULT_ZOOKEEPER_CLIENT_PORT) {
+ util.shutdownMiniCluster()
+ throw new IllegalStateException(
+ s"the mini cluster's ZooKeeper fell back to port $port, so clients
built from a plain " +
+ s"HBaseConfiguration.create() would not find it; is something else
listening on " +
+ s"${HConstants.DEFAULT_ZOOKEEPER_CLIENT_PORT}?")
+ }
+ sys.addShutdownHook(util.shutdownMiniCluster())
+ util
+ }
+
+ /**
+ * Starts the cluster, unless this JVM already has one running. Blocks until
it is ready to serve.
+ */
+ def start(): Unit = {
+ cluster
+ ()
+ }
+}
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index aaa188021..5247260ff 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -324,6 +324,14 @@ object Dependencies {
("org.apache.hadoop" % "hadoop-mapreduce-client-core" % HadoopVersion)
.exclude("ch.qos.reload4j", "reload4j")
.exclude("org.slf4j", "slf4j-reload4j"),
+ // starts an in-JVM HBase for the tests; the shaded flavour is the one
that pairs
+ // with hbase-shaded-client
+ ("org.apache.hbase" % "hbase-shaded-testing-util" % hbaseVersion %
Test)
+ .exclude("log4j", "log4j")
+ .exclude("org.slf4j", "slf4j-log4j12")
+ .exclude("org.apache.logging.log4j", "log4j-slf4j-impl")
+ .exclude("ch.qos.reload4j", "reload4j")
+ .exclude("org.slf4j", "slf4j-reload4j"),
"org.slf4j" % "log4j-over-slf4j" % Slf4jVersion % Test))
}
diff --git a/scripts/hbase-server.sh b/scripts/hbase-server.sh
deleted file mode 100755
index 28a744be6..000000000
--- a/scripts/hbase-server.sh
+++ /dev/null
@@ -1,26 +0,0 @@
-#!/bin/bash
-#
-# 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.
-#
-
-set -x
-
-# The HBase region server registers itself in ZooKeeper under the container
-# hostname "hbase", so the host running the tests must be able to resolve
-# that name to localhost: https://github.com/akka/alpakka/issues/2185
-echo "127.0.0.1 hbase" | sudo tee -a /etc/hosts
-
-docker compose up -d hbase
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
