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:34 UTC
(kudu) 13/14: fixed row result converter for auto incrementing column as well
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 fe9558eb850eb2147da4b207ebb68be2dbd38c87
Author: Marton Greber <gr...@gmail.com>
AuthorDate: Mon Apr 14 11:31:27 2025 +0200
fixed row result converter for auto incrementing column as well
---
java/gradle/dependencies.gradle | 1 +
.../CustomReplicationRowRestultConverter.java | 22 +++++--
.../apache/kudu/replication/ReplicationJob.java | 2 +-
.../kudu/replication/ReplicationJobExecutor.java | 13 ++--
.../apache/kudu/replication/TestReplication.java | 52 ++++++++++++---
.../java/org/apache/kudu/test/ClientTestUtil.java | 76 ++++++++++++++++++++--
6 files changed, 137 insertions(+), 29 deletions(-)
diff --git a/java/gradle/dependencies.gradle b/java/gradle/dependencies.gradle
index 565743f32..289cdc7e8 100755
--- a/java/gradle/dependencies.gradle
+++ b/java/gradle/dependencies.gradle
@@ -96,6 +96,7 @@ libs += [
flinkClients : "org.apache.flink:flink-clients:$versions.flink",
flinkConnectorBase : "org.apache.flink:flink-connector-base:$versions.flink",
flinkCore : "org.apache.flink:flink-core:$versions.flink",
+ flinkApiCommon : "org.apache.flink:flink-api-common:$versions.flink",
flinkDist : "org.apache.flink:flink-dist:$versions.flink",
flinkStreamingJava : "org.apache.flink:flink-streaming-java:$versions.flink",
flinkConnectorKudu : "org.apache.flink:flink-connector-kudu:$versions.flinkConnectorKudu",
diff --git a/java/kudu-replication/src/main/java/org/apache/kudu/replication/CustomReplicationRowRestultConverter.java b/java/kudu-replication/src/main/java/org/apache/kudu/replication/CustomReplicationRowRestultConverter.java
index 8019293c5..0facce546 100644
--- a/java/kudu-replication/src/main/java/org/apache/kudu/replication/CustomReplicationRowRestultConverter.java
+++ b/java/kudu-replication/src/main/java/org/apache/kudu/replication/CustomReplicationRowRestultConverter.java
@@ -16,17 +16,25 @@ public class CustomReplicationRowRestultConverter implements RowResultConverter<
public Row convert(RowResult row) {
Schema schema = row.getColumnProjection();
- Row values = new Row(schema.getColumnCount());
- if (schema.hasIsDeleted()) {
+ int columnCount = schema.hasIsDeleted() ? schema.getColumnCount() - 1 : schema.getColumnCount();
+ Row values = new Row(columnCount);
+
+ int isDeletedIndex = schema.hasIsDeleted() ? schema.getIsDeletedIndex() : -1;
+
+ if (row.hasIsDeleted() && row.isDeleted()) {
values.setKind((RowKind.DELETE));
}
- schema.getColumns().forEach((column) -> {
- String name = column.getName();
- int pos = schema.getColumnIndex(name);
- if (!row.isNull(name)) {
- values.setField(pos, row.getObject(name));
+ schema.getColumns().forEach(column -> {
+ int pos = schema.getColumnIndex(column.getName());
+ // Skip the isDeleted column if it exists
+ if (pos == isDeletedIndex && row.hasIsDeleted()) {
+ return;
+ }
+ if (!row.isNull(pos)) {
+ values.setField(pos, row.getObject(pos));
}
});
+
return values;
}
}
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 5992710e4..c9bb18a7c 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
@@ -57,7 +57,7 @@ public class ReplicationJob {
ReplicationJobExecutor replicationJobExecutor = new ReplicationJobExecutor(config);
- replicationJobExecutor.runJob();
+ replicationJobExecutor.getEnv().execute();
}
}
diff --git a/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJobExecutor.java b/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJobExecutor.java
index ce84527f2..85e86aac5 100644
--- a/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJobExecutor.java
+++ b/java/kudu-replication/src/main/java/org/apache/kudu/replication/ReplicationJobExecutor.java
@@ -19,6 +19,7 @@ package org.apache.kudu.replication;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.connector.kudu.connector.KuduTableInfo;
import org.apache.flink.connector.kudu.connector.converter.RowResultRowConverter;
import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig;
@@ -30,6 +31,8 @@ import org.apache.flink.connector.kudu.source.KuduSourceBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
+import java.time.Duration;
+
/**
* Class for executing the replication actions.
*/
@@ -43,7 +46,7 @@ class ReplicationJobExecutor {
/**
* Executes the actual business logic of the replication.
*/
- public void runJob() throws Exception {
+ public StreamExecutionEnvironment getEnv() throws Exception {
try (ReplicationTableInitializer tableInitializer = new ReplicationTableInitializer(config)) {
tableInitializer.createTableIfNotExists();
@@ -59,16 +62,16 @@ class ReplicationJobExecutor {
KuduSource<Row> kuduSource = new KuduSourceBuilder<Row>()
.setReaderConfig(getReaderConfig())
.setTableInfo(getTableInfo())
- .setRowResultConverter(new RowResultRowConverter())
+ .setRowResultConverter(new CustomReplicationRowRestultConverter())
+ .setBoundedness(Boundedness.CONTINUOUS_UNBOUNDED)
+ .setDiscoveryPeriod(Duration.ofSeconds(2))
.build();
env.fromSource(kuduSource, WatermarkStrategy.noWatermarks(), "KuduSource")
.returns(TypeInformation.of(Row.class))
.sinkTo(kuduSink);
-
- env.execute("Kudu Flink job");
-
+ return env;
}
private KuduWriterConfig getWriterConfig() {
diff --git a/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java b/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
index 9d25adfc8..6a436068b 100644
--- a/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
+++ b/java/kudu-replication/src/test/java/org/apache/kudu/replication/TestReplication.java
@@ -17,23 +17,22 @@
package org.apache.kudu.replication;
-import static org.apache.kudu.test.ClientTestUtil.countRowsInTable;
-import static org.apache.kudu.test.ClientTestUtil.createTableWithOneThousandRows;
+import static org.apache.kudu.test.ClientTestUtil.*;
import static org.apache.kudu.test.KuduTestHarness.DEFAULT_SLEEP;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.util.ArrayList;
import java.util.Arrays;
+import org.apache.flink.api.common.JobID;
+import org.apache.kudu.client.*;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.kudu.client.KuduClient;
-import org.apache.kudu.client.KuduTable;
import org.apache.kudu.test.KuduTestHarness;
public class TestReplication {
@@ -58,8 +57,8 @@ public class TestReplication {
public void testBasicReplication() throws Exception {
//Setup source table
try {
- createTableWithOneThousandRows(
- this.sourceHarness.getAsyncClient(), TABLE_NAME, 10, DEFAULT_SLEEP);
+ createAutoIncrementingTableWithOneThousandRows(
+ this.sourceHarness.getAsyncClient(), TABLE_NAME, 10, DEFAULT_SLEEP);
} catch (Exception e) {
e.printStackTrace();
LOG.error(e.getMessage());
@@ -75,12 +74,47 @@ public class TestReplication {
config.setTableName(TABLE_NAME);
ReplicationJobExecutor executor = new ReplicationJobExecutor(config);
- executor.runJob();
+ JobID jobId = executor.getEnv().executeAsync().getJobID();
KuduTable sinkTable = sinkClient.openTable(TABLE_NAME);
- assertEquals(1000, countRowsInTable(sinkTable));
+ while (1000 != countRowsInTable(sinkTable)) {
+ Thread.sleep(1000);
+ }
+
+ insertAutoIncrementingTableWithOneThousandRows(this.sourceHarness.getAsyncClient(),
+ TABLE_NAME, 1000, 10, DEFAULT_SLEEP);
+
+ KuduTable sourceTable = sourceClient.openTable(config.getTableName());
+ KuduScanner scanner = sourceClient.newScannerBuilder(sourceTable).build();
+ KuduSession session = sourceClient.newSession();
+ ArrayList<RowResult> resultList = new ArrayList<>();
+ while (scanner.hasMoreRows()) {
+ RowResultIterator results = scanner.nextRows();
+ while (results.hasNext()) {
+ RowResult result = results.next();
+ resultList.add(result);
+ }
+ }
+ for (RowResult result : resultList) {
+ Delete delete = sourceTable.newDelete();
+ PartialRow row = delete.getRow();
+ row.addInt("key", result.getInt("key"));
+ row.addLong("auto_incrementing_id", result.getLong("auto_incrementing_id"));
+ row.addInt("column1_i", result.getInt("column1_i"));
+ row.addInt("column2_i", result.getInt("column2_i"));
+ row.addString("column3_s", result.getString("column3_s"));
+ row.addBoolean("column4_b", result.getBoolean("column4_b"));
+ session.apply(delete);
+ session.flush();
+ }
+ sinkTable = sinkClient.openTable(TABLE_NAME);
+ while (0 != countRowsInTable(sinkTable)) {
+ Thread.sleep(1000);
+ System.out.println("Row count: " + countRowsInTable(sinkTable));
+ }
}
+
@Test
public void testSourceTableMissing() {
assertTrue(true);
diff --git a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/ClientTestUtil.java b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/ClientTestUtil.java
index 72b7a2690..3c275cbfb 100644
--- a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/ClientTestUtil.java
+++ b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/ClientTestUtil.java
@@ -272,6 +272,22 @@ public abstract class ClientTestUtil {
return new Schema(columns);
}
+ public static Schema getNonUniqueKeyBasicSchema() {
+ ArrayList<ColumnSchema> columns = new ArrayList<>(5);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).nonUniqueKey(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("column1_i", Type.INT32).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("column2_i", Type.INT32).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("column3_s", Type.STRING)
+ .nullable(true)
+ .desiredBlockSize(4096)
+ .encoding(ColumnSchema.Encoding.DICT_ENCODING)
+ .compressionAlgorithm(ColumnSchema.CompressionAlgorithm.LZ4)
+ .build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("column4_b", Type.BOOL).build());
+ return new Schema(columns);
+ }
+
+
public static CreateTableOptions getBasicCreateTableOptions() {
return new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key"));
}
@@ -333,7 +349,7 @@ public abstract class ClientTestUtil {
public static Upsert createBasicSchemaUpsert(KuduTable table, int key) {
Upsert upsert = table.newUpsert();
PartialRow row = upsert.getRow();
- row.addInt(0, key);
+ row.addInt("key", key);
row.addInt(1, 3);
row.addInt(2, 4);
row.addString(3, "another string");
@@ -361,11 +377,11 @@ public abstract class ClientTestUtil {
public static Insert createBasicSchemaInsert(KuduTable table, int key) {
Insert insert = table.newInsert();
PartialRow row = insert.getRow();
- row.addInt(0, key);
- row.addInt(1, 2);
- row.addInt(2, 3);
- row.addString(3, "a string");
- row.addBoolean(4, true);
+ row.addInt("key", key);
+ row.addInt("column1_i", 2);
+ row.addInt("column2_i", 3);
+ row.addString("column3_s", "a string");
+ row.addBoolean("column4_b", true);
return insert;
}
@@ -426,7 +442,53 @@ public abstract class ClientTestUtil {
return table;
}
- public static Schema createManyVarcharsSchema() {
+ public static KuduTable createAutoIncrementingTableWithOneThousandRows(AsyncKuduClient client,
+ String tableName,
+ final int rowDataSize,
+ final long timeoutMs)
+ throws Exception {
+ final int[] KEYS = new int[] { 250, 500, 750 };
+ final Schema autoIncrementSchema = getNonUniqueKeyBasicSchema();
+ CreateTableOptions builder = getBasicCreateTableOptions();
+ for (int i : KEYS) {
+ PartialRow splitRow = autoIncrementSchema.newPartialRow();
+ splitRow.addInt(0, i);
+ builder.addSplitRow(splitRow);
+ }
+ KuduTable table = client.syncClient().createTable(tableName, autoIncrementSchema, builder);
+ AsyncKuduSession session = client.newSession();
+
+ // create a table with on 4 tablets of 250 rows each
+ for (int key = 0; key < 1000; key++) {
+ Insert insert = createBasicSchemaInsert(table, key);
+ session.apply(insert).join(timeoutMs);
+ }
+ session.close().join(timeoutMs);
+ return table;
+ }
+
+ public static KuduTable insertAutoIncrementingTableWithOneThousandRows(AsyncKuduClient client,
+ String tableName,
+ final int startKey,
+ final int rowDataSize,
+ final long timeoutMs)
+ throws Exception {
+// KuduTable table = client.syncClient().createTable(tableName, autoIncrementSchema, builder);
+ KuduTable table = client.syncClient().openTable(tableName);
+ AsyncKuduSession session = client.newSession();
+
+ // create a table with on 4 tablets of 250 rows each
+ for (int key = startKey; key < startKey + 1000; key++) {
+ Insert insert = createBasicSchemaInsert(table, key);
+ session.apply(insert).join(timeoutMs);
+ }
+ session.close().join(timeoutMs);
+ return table;
+ }
+
+
+
+ public static Schema createManyVarcharsSchema() {
ArrayList<ColumnSchema> columns = new ArrayList<>();
columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.VARCHAR)
.typeAttributes(CharUtil.typeAttributes(10)).key(true).build());