You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mg...@apache.org on 2025/04/14 16:58:23 UTC
(kudu) 02/14: Replace Apache Bahir with flink-connector
This is an automated email from the ASF dual-hosted git repository.
mgreber pushed a commit to branch flink-replication
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit d7d80687ec9214500ace5463756627153ae8e7cb
Author: zchovan <zc...@cloudera.com>
AuthorDate: Tue Jan 7 13:52:09 2025 +0100
Replace Apache Bahir with flink-connector
This commit replaces the Apache Bahir dependency with the Flink Kudu
connector.
Change-Id: Ia1b8d1fb504908d2fc1f693fd3a08a578d72cabe
---
java/build.gradle | 5 +++++
java/gradle/dependencies.gradle | 4 ++--
.../apache/kudu/replication/ReplicationJob.java | 4 ++++
...eplicationHarness.java => TestReplication.java} | 25 ++++++++++++++--------
4 files changed, 27 insertions(+), 11 deletions(-)
diff --git a/java/build.gradle b/java/build.gradle
index ea8f68930..a8eb67c55 100755
--- a/java/build.gradle
+++ b/java/build.gradle
@@ -36,6 +36,11 @@ allprojects {
repositories {
mavenCentral()
mavenLocal()
+ // TODO(zchovan): remove this, once the kudu connector is published upstream
+ // temporarily add downstream nexus
+ maven() {
+ url = uri("https://nexus-private.infra.cloudera.com/nexus/")
+ }
}
// Read the version.txt file to set the project version
diff --git a/java/gradle/dependencies.gradle b/java/gradle/dependencies.gradle
index 354dad276..688bdf8a2 100755
--- a/java/gradle/dependencies.gradle
+++ b/java/gradle/dependencies.gradle
@@ -26,7 +26,7 @@ ext {
versions += [
async : "1.4.1",
- bahir : "1.1.0",
+ bahir : "2.0-SNAPSHOT",
checkstyle : "8.36.1",
clojure : "1.10.3",
clojureToolsCli: "1.0.206",
@@ -97,7 +97,7 @@ libs += [
flinkConnectorBase : "org.apache.flink:flink-connector-base:$versions.flink",
flinkCore : "org.apache.flink:flink-core:$versions.flink",
flinkDist : "org.apache.flink:flink-dist:$versions.flink",
- flinkKuduConnector : "org.apache.bahir:flink-connector-kudu_2.11:$versions.bahir",
+ flinkKuduConnector : "org.apache.flink:flink-connector-kudu:$versions.bahir",
flinkStreamingJava : "org.apache.flink:flink-streaming-java:$versions.flink",
guava : "com.google.guava:guava:$versions.guava",
hadoopClient : "org.apache.hadoop:hadoop-client:$versions.hadoop",
diff --git a/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJob.java b/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJob.java
index 0cb08e181..1c83fd9b2 100644
--- a/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJob.java
+++ b/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJob.java
@@ -22,10 +22,14 @@ import org.apache.flink.api.java.utils.ParameterTool;
import java.util.ArrayList;
import java.util.Collections;
+/**
+ * This class is used to submit Kudu Replication Jobs into a Flink Cluster.
+ */
public class ReplicationJob {
static ReplicationJobConfig config;
public static void main(String[] args) throws Exception {
+ // Read parameters from standard args, then convert them into a ReplicationJobConfig object.
ParameterTool parameters = ParameterTool.fromArgs(args);
config.setSourceMasterAddresses(new ArrayList<>(
diff --git a/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplicationHarness.java b/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
similarity index 69%
rename from java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplicationHarness.java
rename to java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
index 8ee06510a..d62a9209d 100644
--- a/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplicationHarness.java
+++ b/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
@@ -1,7 +1,5 @@
package org.apache.kudu.replication;
-import org.apache.kudu.Schema;
-import org.apache.kudu.client.CreateTableOptions;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.test.KuduTestHarness;
import org.junit.Before;
@@ -12,11 +10,14 @@ import org.slf4j.LoggerFactory;
import java.util.*;
+import static org.apache.kudu.test.ClientTestUtil.createTableWithOneThousandRows;
import static org.apache.kudu.test.ClientTestUtil.getBasicSchema;
+import static org.apache.kudu.test.KuduTestHarness.DEFAULT_SLEEP;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
-public class TestReplicationHarness {
- private static final Logger LOG = LoggerFactory.getLogger(TestReplicationHarness.class);
+public class TestReplication {
+ private static final Logger LOG = LoggerFactory.getLogger(TestReplication.class);
private static final String TABLE_NAME = "replication_test_table";
@Rule
@@ -29,16 +30,22 @@ public class TestReplicationHarness {
@Before
public void setup() {
- CreateTableOptions builder = new CreateTableOptions().setNumReplicas(3);
- Schema basicSchema = getBasicSchema();
-
this.sourceClient = sourceHarness.getClient();
this.sinkClient = sinkHarness.getClient();
-
}
@Test
- public void BenchmarkReplicationScenario() {
+ public void TestBasicReplication() {
+ //Setup source table
+ try {
+ createTableWithOneThousandRows(
+ this.sourceHarness.getAsyncClient(), TABLE_NAME, 32 * 1024, DEFAULT_SLEEP);
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOG.error(e.getMessage());
+ fail(e.getMessage());
+ }
+
ReplicationJobConfig config = new ReplicationJobConfig();
config.setSourceMasterAddresses(Arrays.asList(sourceHarness.getMasterAddressesAsString().split(",")));