I'm trying to connect Apache FlinkSQL with HBase 2.2.0.

I have created an environment with Docker where I have Flink version 1.20 and HBase 2.2.0. I have copied the file flink-sql-connector-hbase-2.2-1.17.2.jar into /opt/flink/lib/. After this, I created the table without any issues.

CREATE TABLE blacklist (
rowkey INT,
 imeis ROW<key STRING, blacklisted BOOLEAN>,
 PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
 'connector' = 'hbase-2.2',
 'table-name' = 'blacklist',
 'zookeeper.quorum' = '127.0.0.1:2181'
);

select * from blacklist;

When I try to do the select I get a NoClassDefFoundError from the hadoop Writable class. The flink-sql-connector-hbase-2.2-1.17.2.jar doesn't contains that class.

Flink SQL> select * from blacklist;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: .apache.hadoop.io.Writable

I've seen that in the test example of the connector, they include all the JARs from the Hadoop distribution (.java). So, I started coping those jars until I got a similar error.

Flink SQL> select * from blacklist;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: .apache.hadoop.conf.Configuration

I checked the jobmanager logs

jobmanager   | 2024-11-18 11:34:59,463 INFO  .apache.flink.runtime.jobmaster.JobMaster                 [] - Running initialization on master for job SELECT `blacklist`.`rowkey`, `blacklist`.`imeis`
jobmanager   | FROM `default_catalog`.`default_database`.`blacklist` AS `blacklist` (7e630a2c6869ee955fbb7fdc13ee31e8).
jobmanager   | 2024-11-18 11:34:59,466 INFO  .apache.flink.runtime.dispatcher.StandaloneDispatcher     [] - Job 7e630a2c6869ee955fbb7fdc13ee31e8 reached terminal state FAILED.
jobmanager   | .apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
jobmanager   |  at .apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97)
jobmanager   |  at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
jobmanager   |  at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
jobmanager   |  at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
jobmanager   |  at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
jobmanager   |  at .apache.flink.util.MdcUtils.lambda$wrapRunnable$1(MdcUtils.java:64)
jobmanager   |  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
jobmanager   |  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
jobmanager   |  at java.base/java.lang.Thread.run(Unknown Source)
jobmanager   | Caused by: java.util.concurrent.CompletionException: java.lang.RuntimeException: .apache.flink.runtime.client.JobExecutionException: Cannot initialize task 'Source: blacklist[7] -> ConstraintEnforcer[8] -> Sink: Collect table sink': Loading the input/output formats failed:
jobmanager   |  at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
jobmanager   |  at java.base/java.util.concurrent.CompletableFuturepleteThrowable(Unknown Source)
jobmanager   |  ... 5 more
jobmanager   | Caused by: java.lang.RuntimeException: .apache.flink.runtime.client.JobExecutionException: Cannot initialize task 'Source: blacklist[7] -> ConstraintEnforcer[8] -> Sink: Collect table sink': Loading the input/output formats failed:
jobmanager   |  at .apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321)
jobmanager   |  at .apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:114)
jobmanager   |  ... 5 more
jobmanager   | Caused by: .apache.flink.runtime.client.JobExecutionException: Cannot initialize task 'Source: blacklist[7] -> ConstraintEnforcer[8] -> Sink: Collect table sink': Loading the input/output formats failed:
jobmanager   |  at .apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:207)
jobmanager   |  at .apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:173)
jobmanager   |  at .apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:382)
jobmanager   |  at .apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:225)
jobmanager   |  at .apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:142)
jobmanager   |  at .apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:162)
jobmanager   |  at .apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:121)
jobmanager   |  at .apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:406)
jobmanager   |  at .apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:383)
jobmanager   |  at .apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:128)
jobmanager   |  at .apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:100)
jobmanager   |  at .apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
jobmanager   |  ... 5 more
jobmanager   | Caused by: java.lang.Exception: Loading the input/output formats failed:
jobmanager   |  at .apache.flink.runtime.jobgraph.InputOutputFormatVertex.initInputOutputformatContainer(InputOutputFormatVertex.java:189)
jobmanager   |  at .apache.flink.runtime.jobgraph.InputOutputFormatVertex.initializeOnMaster(InputOutputFormatVertex.java:59)
jobmanager   |  at .apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:198)
jobmanager   |  ... 16 more
jobmanager   | Caused by: java.lang.RuntimeException: Deserializing the input/output formats failed: /apache/hadoop/conf/Configuration
jobmanager   |  at .apache.flink.runtime.jobgraph.InputOutputFormatContainer.<init>(InputOutputFormatContainer.java:69)
jobmanager   |  at .apache.flink.runtime.jobgraph.InputOutputFormatVertex.initInputOutputformatContainer(InputOutputFormatVertex.java:185)
jobmanager   |  ... 18 more
jobmanager   | Caused by: java.lang.NoClassDefFoundError: /apache/hadoop/conf/Configuration
jobmanager   |  at java.base/java.lang.Class.getDeclaredMethods0(Native Method)
jobmanager   |  at java.base/java.lang.Class.privateGetDeclaredMethods(Unknown Source)
jobmanager   |  at java.base/java.lang.Class.getDeclaredMethod(Unknown Source)

The connector documentation really only talks about copying the library to Flink's lib directory and that's it. Has anyone had similar issues?