This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch flink-iotdb-table-connector in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit 1728355fe0f9490a901c0f06326b4320af7654e0 Author: shuwenwei <[email protected]> AuthorDate: Sat Oct 10 12:18:03 2026 +0800 fix: fix CDC subscription reader column indexing and polling Use one-based source column indexes for the subscription result set, keep the reader available so it keeps polling, and align IoTDB/thrift/tsfile dependency versions. --- .../flink/source/cdc/SubscriptionDataIterator.java | 4 +- .../source/cdc/IoTDBSubscriptionSourceReader.java | 3 +- .../flink/IoTDBRelationalLocalQueryManual.java | 147 ++++++++++++++++++++- pom.xml | 14 +- 4 files changed, 158 insertions(+), 10 deletions(-) diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java index 3e36c4e..6e29bfb 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java @@ -59,9 +59,11 @@ public class SubscriptionDataIterator implements IoTDBDataIterator { private static int[] resolveSourceIndexes( List<String> availableColumns, List<String> projectedColumns) { + // The underlying SubscriptionResultSet is 1-based (see AbstractResultSet.getField), so store + // one-based source column indexes while the projected position stays 0-based. Map<String, Integer> indexByName = new HashMap<>(availableColumns.size()); for (int i = 0; i < availableColumns.size(); i++) { - indexByName.put(normalize(availableColumns.get(i)), i); + indexByName.put(normalize(availableColumns.get(i)), i + 1); } int[] indexes = new int[projectedColumns.size()]; for (int i = 0; i < projectedColumns.size(); i++) { diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java index 8673275..75d865f 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java @@ -102,7 +102,8 @@ public class IoTDBSubscriptionSourceReader<OUT> } fetchBatch(); if (buffer.isEmpty()) { - markUnavailable(); + // Stay available: the next pollNext call blocks in consumer.poll(pollTimeout) and keeps the + // reader polling the subscription. Marking unavailable here would never be resumed. return InputStatus.NOTHING_AVAILABLE; } } diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java index 6bff06a..45a290e 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java @@ -19,6 +19,12 @@ package org.apache.iotdb.relational.flink; +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.session.TableSessionBuilder; +import org.apache.iotdb.session.subscription.ISubscriptionTableSession; +import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder; +import org.apache.iotdb.session.subscription.model.Subscription; + import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; @@ -27,12 +33,16 @@ import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.types.Row; import org.apache.flink.util.CloseableIterator; +import java.util.Arrays; +import java.util.Set; + /** * Temporary manual verification class. This is intentionally kept in test sources and should not be * committed as a production test. * * <p>The flow is write first and then query: rows are inserted through the connector and read back - * with a SELECT, followed by a lookup join (both sync and async) executed as Flink SQL. + * with a SELECT, followed by a lookup join (both sync and async) and a subscription-backed CDC + * read, all executed as Flink SQL. * * <p>Run with system properties such as: * @@ -86,6 +96,104 @@ public class IoTDBRelationalLocalQueryManual { // 3) Lookup: run a temporal join through SQL for both sync and async lookup. verifyLookupJoin(nodeUrls, user, password, database, table, false); verifyLookupJoin(nodeUrls, user, password, database, table, true); + + // 4) CDC: read the table through the subscription-backed CDC source. + verifyCdc(nodeUrls, user, password, database, table); + } + + /** + * Flushes the database so the written rows are sealed into TsFiles and thus visible to the + * subscription snapshot phase (mirrors the official subscription example, which flushes before + * subscribing). + */ + private static void flushDatabase( + String nodeUrls, String user, String password, String database) throws Exception { + System.out.println("\n=== FLUSH database '" + database + "' ==="); + try (ITableSession session = + new TableSessionBuilder() + .nodeUrls(Arrays.asList(nodeUrls.split(","))) + .username(user) + .password(password) + .database(database) + .build()) { + session.executeNonQueryStatement("flush"); + } + long waitMs = Long.getLong("iotdb.cdc.flush.wait.ms", 3000L); + if (waitMs > 0) { + System.out.println("Waiting " + waitMs + " ms for the flush to seal data..."); + Thread.sleep(waitMs); + } + System.out.println("Flush finished."); + } + + /** + * Drops any existing subscription and topic for this (database, table) so each run starts from a + * clean subscription state. The names mirror the ones derived by {@code IoTDBOptions}. + */ + private static void resetCdcSubscription( + String nodeUrls, String user, String password, String database, String table) { + String topic = "flink_iotdb_table_" + sanitize(database) + "_" + sanitize(table); + String[] hostPort = nodeUrls.split(",")[0].split(":"); + System.out.println("\n=== CDC RESET: drop subscriptions and topic '" + topic + "' ==="); + + try (ISubscriptionTableSession session = + new SubscriptionTableSessionBuilder() + .host(hostPort[0]) + .port(Integer.parseInt(hostPort[1])) + .username(user) + .password(password) + .build()) { + session.open(); + try { + for (Subscription subscription : session.getSubscriptions(topic)) { + System.out.println(" drop subscription " + subscription.getSubscriptionId()); + session.dropSubscriptionIfExists(subscription.getSubscriptionId()); + } + } catch (Exception e) { + System.out.println(" list/drop subscriptions failed: " + e.getMessage()); + } + session.dropTopicIfExists(topic); + System.out.println(" reset done."); + } catch (Exception e) { + System.out.println(" CDC reset skipped/failed: " + e.getMessage()); + } + } + + private static String sanitize(String value) { + return value == null ? "" : value.replaceAll("[^A-Za-z0-9_]", "_"); + } + + /** + * Reads the table through the CDC source. The database is flushed first so the written rows are + * visible to the subscription snapshot phase, then the topic/subscription is reset and recreated + * in {@code initial} mode (full + incremental); a LIMIT bounds the otherwise unbounded stream. + */ + private static void verifyCdc( + String nodeUrls, String user, String password, String database, String table) + throws Exception { + flushDatabase(nodeUrls, user, password, database); + resetCdcSubscription(nodeUrls, user, password, database, table); + + StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); + environment.setParallelism(1); + StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment); + + String flinkTable = "iotdb_cdc"; + tableEnvironment.executeSql("DROP TABLE IF EXISTS " + flinkTable); + String ddl = buildCdcDdl(flinkTable, nodeUrls, user, password, database, table); + System.out.println("\n=== CDC DDL ===\n" + ddl); + tableEnvironment.executeSql(ddl); + + System.out.println("\n=== CDC (snapshot mode) ==="); + String sql = "SELECT * FROM " + flinkTable + ""; + System.out.println(sql); + TableResult result = tableEnvironment.executeSql(sql); + try (CloseableIterator<Row> iterator = result.collect()) { + while (iterator.hasNext()) { + System.out.println(iterator.next()); + } + } + System.out.println("CDC read finished."); } private static void verifyLookupJoin( @@ -164,6 +272,43 @@ public class IoTDBRelationalLocalQueryManual { return ddl.toString(); } + private static String buildCdcDdl( + String flinkTable, + String nodeUrls, + String user, + String password, + String database, + String table) { + return "CREATE TABLE " + + flinkTable + + " (\n" + + " `time` TIMESTAMP(3),\n" + + " `device_id` STRING,\n" + + " `temperature` DOUBLE\n" + + ") WITH (\n" + + " 'connector' = 'iotdb-relational',\n" + + " 'iotdb.node-urls' = '" + + nodeUrls + + "',\n" + + " 'iotdb.user' = '" + + user + + "',\n" + + " 'iotdb.password' = '" + + password + + "',\n" + + " 'iotdb.database' = '" + + database + + "',\n" + + " 'iotdb.table' = '" + + table + + "',\n" + + " 'iotdb.time-column' = 'time',\n" + + " 'iotdb.tag-columns' = 'device_id',\n" + + " 'iotdb.scan.mode' = 'cdc',\n" + + " 'iotdb.cdc.mode' = 'initial'\n" + + ")"; + } + private static void execute(TableEnvironment tableEnvironment, String sql) throws Exception { System.out.println("\n=== WRITE ==="); System.out.println(sql); diff --git a/pom.xml b/pom.xml index 707e985..07c88c3 100644 --- a/pom.xml +++ b/pom.xml @@ -90,8 +90,8 @@ This is the version of the thrift binary, that we release separately from here: https://github.com/apache/iotdb-bin-resources/tree/main/iotdb-tools-thrift --> - <iotdb-tools-thrift.version>0.14.1.0</iotdb-tools-thrift.version> - <iotdb.version>2.0.7-SNAPSHOT</iotdb.version> + <iotdb-tools-thrift.version>0.23.0.0</iotdb-tools-thrift.version> + <iotdb.version>2.0.11-SNAPSHOT</iotdb.version> <jackson.version>2.16.2</jackson.version> <!-- This is the last version to support the javax namespace --> <jakarta.servlet-api.version>4.0.4</jakarta.servlet-api.version> @@ -114,7 +114,7 @@ <kafka.version>2.8.2</kafka.version> <!-- This was the last version to support Java 8 --> <logback.version>1.3.15</logback.version> - <lz4-java.version>1.8.0</lz4-java.version> + <lz4-java.version>1.11.1</lz4-java.version> <maven.assembly.version>3.6.0</maven.assembly.version> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> @@ -152,7 +152,7 @@ <!-- Newer version requires refactoring of testsuite --> <scalatest.version>3.0.9</scalatest.version> <slf4j.version>2.0.9</slf4j.version> - <snappy-java.version>1.1.10.4</snappy-java.version> + <snappy-java.version>1.1.10.5</snappy-java.version> <sonar.coverage.jacoco.xmlReportPaths>target/jacoco-merged-reports/jacoco.xml</sonar.coverage.jacoco.xmlReportPaths> <!-- Exclude all generated code --> <sonar.exclusions>**/generated-sources</sonar.exclusions> @@ -181,13 +181,13 @@ All versions between 0.17.0 and 0.14.1 have know vulnerabilities, so for now we'll stay at 0.14.1. --> - <thrift.version>0.14.1</thrift.version> + <thrift.version>0.24.0</thrift.version> <!-- This was the last version to support Java 8 --> <tomcat.version>9.0.86</tomcat.version> - <tsfile.version>2.4.0</tsfile.version> + <tsfile.version>2.4.1-260915-SNAPSHOT</tsfile.version> <xz.version>1.9</xz.version> <zeppelin.version>0.11.1</zeppelin.version> - <zstd-jni.version>1.5.5-5</zstd-jni.version> + <zstd-jni.version>1.5.6-3</zstd-jni.version> </properties> <!-- if we claim dependencies in dependencyManagement, then we do not claim
