This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.4.x
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/1.4.x by this push:
new c27f8daa8 HBase: re-enable integration tests in CI on 1.4.x (#1910)
c27f8daa8 is described below
commit c27f8daa8822af6eaa3217e0b64f469c157d12c0
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Sep 5 16:50:13 2026 +0100
HBase: re-enable integration tests in CI on 1.4.x (#1910)
Motivation:
The hbase CI job has been disabled since 2020 (inherited from alpakka),
so the HBase connector has no integration coverage on 1.4.x. The root
cause in akka/alpakka#2185 was that the harisekhon/hbase container's
region server registers itself in ZooKeeper under the hostname "hbase",
which the runner cannot resolve - all tests then time out. The 2020
image rebuild also moved the region server to the default ports.
Modification:
Backport of the CI parts of #1895, without the docker image or hbase
client upgrades (the 1.4.x branch keeps harisekhon/hbase:1.4 and
hbase-shaded-client 1.4.14, so the scala-steward pins and the 1.x-API
code also stay):
- Add scripts/hbase-server.sh (copied from #1895; same pattern as
scripts/ftp-servers.sh, since $PRE_CMD cannot contain shell
operators): appends "127.0.0.1 hbase" to /etc/hosts and starts the
hbase docker-compose service; re-enable the hbase entry in the CI
connectors matrix.
- Publish the HBase-default ports 16020/16030 instead of the stale
16201/16301.
- Add the directional flow tests for append, increment, delete and
complex converters from #1895, adapted to the 1.4.x test style
(JDK 8 idioms, Append.add instead of the 2.x Append.addColumn).
Result:
The HBase connector has working CI integration tests again on 1.4.x,
unchanged docker image and client versions.
Tests:
- sbt "hbase/Test/compile" and sbt "hbase/javafmtCheckAll" - clean
- scalafmt --mode diff-ref=upstream/1.4.x - clean
- Integration runs in the re-enabled connectors (hbase) CI job; this
exact configuration (1.4 image, 1.4.14 client, ports 16020/16030,
hosts-file script) passed CI on main at apache/pekko-connectors@e6e7963ba
before #1895 upgraded the image/client. Local integration not run
(requires sudo to edit /etc/hosts).
References:
Refs #1895, Refs #61, Refs https://github.com/akka/alpakka/issues/2185
---
.github/workflows/check-build-test.yml | 3 +-
docker-compose.yml | 4 +-
.../src/test/java/docs/javadsl/HBaseStageTest.java | 112 +++++++++++++++++++++
.../test/scala/docs/scaladsl/HBaseStageSpec.scala | 68 +++++++++++++
scripts/hbase-server.sh | 26 +++++
5 files changed, 209 insertions(+), 4 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index 43a9edeba..71988eca2 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -116,8 +116,7 @@ jobs:
- { connector: google-cloud-storage }
- { connector: google-common }
- { connector: google-fcm }
- # hbase disabled until we resolve why new docker image fails our
build: https://github.com/akka/alpakka/issues/2185
- # - { connector: hbase, pre_cmd: 'docker
compose up -d hbase' }
+ - { connector: hbase, pre_cmd:
'./scripts/hbase-server.sh' }
- { connector: hdfs, pre_cmd: 'file
${HOME}/.cache/coursier/v1/https/repo1.maven.org/maven2/org/typelevel/cats-kernel_2.13/2.0.0/cats-kernel_2.13-2.0.0.jar'
}
- { connector: huawei-push-kit }
- { connector: influxdb, pre_cmd: 'docker
compose up -d influxdb' }
diff --git a/docker-compose.yml b/docker-compose.yml
index 7ad3fc0cd..3699b057f 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -115,8 +115,8 @@ services:
- 2181:2181
- 16000:16000
- 16010:16010
- - 16201:16201
- - 16301:16301
+ - 16020:16020
+ - 16030:16030
ibmmq:
image: ibmcom/mq:latest
environment:
diff --git a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
index eec183399..2625183cc 100644
--- a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
+++ b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
@@ -28,12 +28,14 @@ import org.apache.pekko.testkit.javadsl.TestKit;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import java.io.UnsupportedEncodingException;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -221,6 +223,116 @@ public class HBaseStageTest {
assertEquals(1, f.toCompletableFuture().get().size());
}
+
+ @Test
+ public void appendThroughFlow() throws Exception {
+ HTableSettings<Person> appendSettings =
mutationTableSettings(appendHBaseConverter);
+
+ // unique row per run: the sbt cross-build reruns this suite against the
same HBase instance
+ int id = randomRowId();
+ CompletionStage<Done> f =
+ Source.from(Arrays.asList(new Person(id, "-a"), new Person(id, "-b")))
+ .via(HTableStage.flow(appendSettings))
+ .runWith(Sink.ignore(), system);
+ assertEquals(Done.getInstance(), f.toCompletableFuture().get(5,
TimeUnit.SECONDS));
+
+ List<Result> results = readRow(appendSettings, id);
+ assertEquals(1, results.size());
+ assertEquals(
+ "-a-b",
+ new String(
+ results.get(0).getValue(bytes("info"), bytes("aliases")),
StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void incrementThroughFlow() throws Exception {
+ HTableSettings<Person> incrementSettings =
mutationTableSettings(incrementHBaseConverter);
+
+ int id = randomRowId();
+ CompletionStage<Done> f =
+ Source.from(
+ Arrays.asList(new Person(id, "inc"), new Person(id, "inc"),
new Person(id, "inc")))
+ .via(HTableStage.flow(incrementSettings))
+ .runWith(Sink.ignore(), system);
+ assertEquals(Done.getInstance(), f.toCompletableFuture().get(5,
TimeUnit.SECONDS));
+
+ List<Result> results = readRow(incrementSettings, id);
+ assertEquals(1, results.size());
+ assertEquals(
+ 3L, Bytes.toLong(results.get(0).getValue(bytes("info"),
bytes("numberOfChanges"))));
+ }
+
+ @Test
+ public void deleteThroughFlow() throws Exception {
+ HTableSettings<Person> putSettings = mutationTableSettings(hBaseConverter);
+ int id = randomRowId();
+ CompletionStage<Done> put =
+ Source.single(new Person(id, "to be deleted"))
+ .runWith(HTableStage.sink(putSettings), system);
+ assertEquals(Done.getInstance(), put.toCompletableFuture().get(5,
TimeUnit.SECONDS));
+ assertEquals(1, readRow(putSettings, id).size());
+
+ HTableSettings<Person> deleteSettings =
mutationTableSettings(deleteHBaseConverter);
+ CompletionStage<Done> delete =
+ Source.single(new Person(id, "to be deleted"))
+ .via(HTableStage.flow(deleteSettings))
+ .runWith(Sink.ignore(), system);
+ assertEquals(Done.getInstance(), delete.toCompletableFuture().get(5,
TimeUnit.SECONDS));
+
+ assertEquals(0, readRow(putSettings, id).size());
+ }
+
+ @Test
+ public void complexConverterThroughFlow() throws Exception {
+ HTableSettings<Person> complexSettings =
mutationTableSettings(complexHBaseConverter);
+
+ int mixedId = randomRowId();
+ int deletedId = randomRowId();
+ CompletionStage<List<Person>> f =
+ Source.from(
+ Arrays.asList(
+ new Person(0, "skipped"),
+ new Person(mixedId, "mixed"),
+ new Person(deletedId, "")))
+ .via(HTableStage.flow(complexSettings))
+ .runWith(Sink.seq(), system);
+ assertEquals(3, f.toCompletableFuture().get(5, TimeUnit.SECONDS).size());
+
+ List<Result> results = readRow(complexSettings, mixedId);
+ assertEquals(1, results.size());
+ assertEquals(
+ "mixed",
+ new String(results.get(0).getValue(bytes("info"), bytes("name")),
StandardCharsets.UTF_8));
+ assertEquals(
+ 1L, Bytes.toLong(results.get(0).getValue(bytes("info"),
bytes("numberOfChanges"))));
+ assertEquals(0, readRow(complexSettings, 0).size());
+ assertEquals(0, readRow(complexSettings, deletedId).size());
+ }
+
+ private static int randomRowId() {
+ return 1000
+ +
java.util.concurrent.ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE -
1000);
+ }
+
+ private static byte[] bytes(String s) {
+ return s.getBytes(StandardCharsets.UTF_8);
+ }
+
+ private HTableSettings<Person> mutationTableSettings(Function<Person,
List<Mutation>> converter) {
+ return HTableSettings.create(
+ HBaseConfiguration.create(),
+ TableName.valueOf("person3"),
+ Collections.singletonList("info"),
+ converter);
+ }
+
+ private List<Result> readRow(HTableSettings<Person> tableSettings, int id)
throws Exception {
+ Scan scan = new Scan(new Get(String.format("id_%d",
id).getBytes(StandardCharsets.UTF_8)));
+ return HTableStage.source(scan, tableSettings)
+ .runWith(Sink.seq(), system)
+ .toCompletableFuture()
+ .get(5, TimeUnit.SECONDS);
+ }
}
class Person {
diff --git a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
index a6a867107..b60365cdf 100644
--- a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
+++ b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
@@ -30,6 +30,7 @@ import org.scalatest.BeforeAndAfterAll
import scala.collection.immutable
import scala.concurrent.duration._
import scala.language.implicitConversions
+import scala.util.Random
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
@@ -145,8 +146,75 @@ class HBaseStageSpec
f.futureValue.size shouldBe 1
}
+
+ "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
+ val id = randomRowId()
+ val f = Source(List(Person(id, "-a"), Person(id, "-b")))
+ .via(HTableStage.flow(appendSettings))
+ .runWith(Sink.ignore)
+ f.futureValue shouldBe Done
+
+ val results = readRow(id).futureValue
+ results should have size 1
+ Bytes.toString(results.head.getValue("info", "aliases")) shouldBe "-a-b"
+ }
+
+ "increment a cell through a flow" in {
+ val incrementSettings =
tableSettings.withConverter(incrementHBaseConverter)
+
+ val id = randomRowId()
+ val f = Source(List.fill(3)(Person(id, "increment")))
+ .via(HTableStage.flow(incrementSettings))
+ .runWith(Sink.ignore)
+ f.futureValue shouldBe Done
+
+ val results = readRow(id).futureValue
+ results should have size 1
+ Bytes.toLong(results.head.getValue("info", "numberOfChanges")) shouldBe
3L
+ }
+
+ "delete a row through a flow" in {
+ val id = randomRowId()
+ Source.single(Person(id, "to be
deleted")).runWith(HTableStage.sink(tableSettings)).futureValue shouldBe Done
+ readRow(id).futureValue should have size 1
+
+ val deleteSettings = tableSettings.withConverter(deleteHBaseConverter)
+ Source
+ .single(Person(id, "to be deleted"))
+ .via(HTableStage.flow(deleteSettings))
+ .runWith(Sink.ignore)
+ .futureValue shouldBe Done
+
+ readRow(id).futureValue shouldBe empty
+ }
+
+ "apply a converter that emits multiple or no mutations" in {
+ val complexSettings =
tableSettings.withConverter(mutationsHBaseConverter)
+
+ val mixedId = randomRowId()
+ val deletedId = randomRowId()
+ val f = Source(List(Person(0, "skipped"), Person(mixedId, "mixed"),
Person(deletedId, "")))
+ .via(HTableStage.flow(complexSettings))
+ .runWith(Sink.seq)
+ f.futureValue should have size 3
+
+ val results = readRow(mixedId).futureValue
+ results should have size 1
+ Bytes.toString(results.head.getValue("info", "name")) shouldBe "mixed"
+ Bytes.toLong(results.head.getValue("info", "numberOfChanges")) shouldBe
1L
+ readRow(0).futureValue shouldBe empty
+ readRow(deletedId).futureValue shouldBe empty
+ }
}
+ private def randomRowId(): Int = 1000 + Random.nextInt(Int.MaxValue - 1000)
+
+ private def readRow(id: Int) =
+ HTableStage.source(new Scan(new Get(Bytes.toBytes(s"id_$id"))),
tableSettings).runWith(Sink.seq)
+
override def afterAll(): Unit =
TestKit.shutdownActorSystem(system)
}
diff --git a/scripts/hbase-server.sh b/scripts/hbase-server.sh
new file mode 100755
index 000000000..28a744be6
--- /dev/null
+++ b/scripts/hbase-server.sh
@@ -0,0 +1,26 @@
+#!/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]
