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 188afd20b HBase: re-enable integration tests in CI (#1895)
188afd20b is described below
commit 188afd20b3de2c26ffe9c3c462b1cf19a3f877ec
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 4 09:47:14 2026 +0100
HBase: re-enable integration tests in CI (#1895)
* HBase: re-enable integration tests in CI
* publish hbase 1.4 default ports 16020/16030 instead of stale 16201/16301
* add license header to hbase-server.sh
* try harisekhon/hbase:2.1 image
* upgrade hbase client to 2.6.6, unpin hbase/hadoop from scala-steward
* add directional tests for append, increment, delete and complex converters
* use unique row ids so hbase tests survive cross-build reruns
* fix undefined row ids in java complex converter test
---
.github/workflows/check-build-test.yml | 3 +-
.scala-steward.conf | 3 -
docker-compose.yml | 6 +-
.../connectors/hbase/impl/HBaseCapabilities.scala | 16 ++-
.../src/test/java/docs/javadsl/HBaseStageTest.java | 109 ++++++++++++++++++++-
.../test/scala/docs/scaladsl/HBaseStageSpec.scala | 70 ++++++++++++-
project/Dependencies.scala | 2 +-
scripts/hbase-server.sh | 26 +++++
8 files changed, 219 insertions(+), 16 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index 482e263c1..27eee0cb5 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -120,8 +120,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/.scala-steward.conf b/.scala-steward.conf
index d1ba2d8ce..dad5d4976 100644
--- a/.scala-steward.conf
+++ b/.scala-steward.conf
@@ -16,9 +16,6 @@ updates.pin = [
updates.ignore = [
# Manage Pekko upgrades explicitly
{ groupId = "org.apache.pekko" }
- # https://github.com/apache/pekko-connectors/issues/61
- { groupId = "org.apache.hbase" }
- { groupId = "org.apache.hadoop" }
# Avoid scala-steward opening multiple PRs for Jackson version updates,
# as they are managed by a single variable in our build
{ groupId = "com.fasterxml.jackson.core" }
diff --git a/docker-compose.yml b/docker-compose.yml
index 0a13144dd..bd3a2e3df 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -108,14 +108,14 @@ services:
python subscriber.py pekko-connectors create testTopic testSubscription
"
hbase:
- image: harisekhon/hbase:1.4
+ image: harisekhon/hbase:2.1
hostname: hbase
ports:
- 2181:2181
- 16000:16000
- 16010:16010
- - 16201:16201
- - 16301:16301
+ - 16020:16020
+ - 16030:16030
ibmmq:
image: icr.io/ibm-messaging/mq:9.4.5.1-r1
environment:
diff --git
a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseCapabilities.scala
b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseCapabilities.scala
index b2cea1ebb..e1af465da 100644
---
a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseCapabilities.scala
+++
b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseCapabilities.scala
@@ -16,8 +16,14 @@ package org.apache.pekko.stream.connectors.hbase.impl
import java.io.Closeable
import org.apache.pekko.stream.stage.StageLogging
-import org.apache.hadoop.hbase.{ HColumnDescriptor, HTableDescriptor,
TableName }
-import org.apache.hadoop.hbase.client.{ Connection, ConnectionFactory, Table }
+import org.apache.hadoop.hbase.TableName
+import org.apache.hadoop.hbase.client.{
+ ColumnFamilyDescriptorBuilder,
+ Connection,
+ ConnectionFactory,
+ Table,
+ TableDescriptorBuilder
+}
import org.apache.hadoop.conf.Configuration
import scala.concurrent.duration.DurationInt
@@ -58,11 +64,11 @@ private[impl] trait HBaseCapabilities { this: StageLogging
=>
if (admin.isTableAvailable(tableName))
connection.getTable(tableName)
else {
- val tableDescriptor: HTableDescriptor = new HTableDescriptor(tableName)
+ val tableDescriptorBuilder =
TableDescriptorBuilder.newBuilder(tableName)
columnFamilies.foreach { cf =>
- tableDescriptor.addFamily(new HColumnDescriptor(cf))
+
tableDescriptorBuilder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(cf))
}
- admin.createTable(tableDescriptor)
+ admin.createTable(tableDescriptorBuilder.build())
log.info("Table {} created with cfs: {}.", tableName, columnFamilies)
connection.getTable(tableName)
}
diff --git a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
index 660e9841e..0e23b7bae 100644
--- a/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
+++ b/hbase/src/test/java/docs/javadsl/HBaseStageTest.java
@@ -26,6 +26,7 @@ import java.util.function.Function;
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.apache.pekko.Done;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
@@ -75,7 +76,7 @@ public class HBaseStageTest {
Function<Person, List<Mutation>> appendHBaseConverter =
person -> {
Append append = new
Append("id_%d".formatted(person.id).getBytes(StandardCharsets.UTF_8));
- append.add(
+ append.addColumn(
"info".getBytes(StandardCharsets.UTF_8),
"aliases".getBytes(StandardCharsets.UTF_8),
person.name.getBytes(StandardCharsets.UTF_8));
@@ -204,6 +205,112 @@ 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(List.of(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(List.of(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(
+ List.of(
+ 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"),
List.of("info"), converter);
+ }
+
+ private List<Result> readRow(HTableSettings<Person> tableSettings, int id)
throws Exception {
+ Scan scan = new Scan(new
Get("id_%d".formatted(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..7459f88bc 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
@@ -59,7 +60,7 @@ class HBaseStageSpec
val appendHBaseConverter: Person => immutable.Seq[Mutation] = { person =>
// Append to a cell
val append = new Append(s"id_${person.id}")
- append.add("info", "aliases", person.name)
+ append.addColumn("info", "aliases", person.name)
List(append)
}
// #create-converter-append
@@ -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/project/Dependencies.scala b/project/Dependencies.scala
index 267236fe8..4d2879508 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -309,7 +309,7 @@ object Dependencies {
"io.specto" % "hoverfly-java" % hoverflyVersion % Test) ++ Mockito)
val HBase = {
- val hbaseVersion = "1.4.14"
+ val hbaseVersion = "2.6.6"
Seq(
libraryDependencies ++= Seq(
("org.apache.hbase" % "hbase-shaded-client" % hbaseVersion)
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]
