This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new b5f65fb9 [Improve] Refine Doris source logging and offset validation
(#692)
b5f65fb9 is described below
commit b5f65fb98efbfa1efafb553b6bec9bfab853754f
Author: wudi <[email protected]>
AuthorDate: Fri Aug 28 17:25:13 2026 +0800
[Improve] Refine Doris source logging and offset validation (#692)
Report the number of records read when a Flight SQL split reaches end of
stream.
Move routine Doris FE HTTP request logging from INFO to DEBUG.
Keep incremental SQL logging at DEBUG to avoid repetitive INFO output.
Validate the MySQL JDBC driver, connectivity, and configured offset table
before creating an incremental Doris source.
Fail job submission early when offset persistence cannot be initialized
instead of discovering the failure after snapshot processing.
---
.../org/apache/doris/flink/rest/RestService.java | 2 +-
.../source/reader/DorisFlightValueReader.java | 12 +++++-
.../flink/source/reader/DorisOffsetPublisher.java | 28 +++++++++++++-
.../source/reader/DorisOffsetPublisherTest.java | 43 ++++++++++++++++++++++
.../org/apache/doris/flink/source/DorisSource.java | 9 +++++
.../doris/flink/source/DorisSourceOptionsTest.java | 42 ++++++++++++++++-----
.../org/apache/doris/flink/source/DorisSource.java | 9 +++++
.../apache/doris/flink/sink/DorisSinkITCase.java | 27 +++++++++-----
8 files changed, 148 insertions(+), 24 deletions(-)
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
index ca00ec8b..945c8e16 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
@@ -262,7 +262,7 @@ public class RestService implements Serializable {
.build();
request.setConfig(requestConfig);
- logger.info(
+ logger.debug(
"Send request to Doris FE '{}' with user '{}'.",
request.getURI(),
options.getUsername());
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
index 32148024..63abef14 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
@@ -80,6 +80,7 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
AdbcStatement.QueryResult queryResult;
protected ArrowReader arrowReader;
protected AtomicBoolean eos = new AtomicBoolean(false);
+ private long readRowCount;
public DorisFlightValueReader(
DorisSourceSplit split, DorisOptions options, DorisReadOptions
readOptions) {
@@ -297,7 +298,12 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
}
if (!eos.get()) {
eos.set(!arrowReader.loadNextBatch());
- if (!eos.get()) {
+ if (eos.get()) {
+ LOG.info(
+ "Flight SQL scan finished, split: {}, records:
{}",
+ split.splitId(),
+ readRowCount);
+ } else {
rowBatch =
new RowBatch(
arrowReader,
@@ -330,7 +336,9 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
LOG.error(SHOULD_NOT_HAPPEN_MESSAGE);
throw new ShouldNeverHappenException();
}
- return rowBatch.nextSourceRecord();
+ DorisSourceRecord record = rowBatch.nextSourceRecord();
+ readRowCount++;
+ return record;
}
@Override
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
index d4c2c5c7..76d6f351 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
@@ -20,12 +20,16 @@ import
org.apache.doris.flink.connection.JdbcConnectionProvider;
import org.apache.doris.flink.exception.DorisRuntimeException;
import java.sql.Connection;
+import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.util.function.Consumer;
/** Publishes completed-checkpoint offsets through Doris JDBC. */
public class DorisOffsetPublisher implements AutoCloseable {
private final JdbcConnectionProvider connectionProvider;
+ private final String database;
+ private final String table;
private final String insertSql;
private final String consumerId;
@@ -36,16 +40,36 @@ public class DorisOffsetPublisher implements AutoCloseable {
if (tableParts.length != 2 || tableParts[0].isEmpty() ||
tableParts[1].isEmpty()) {
throw new IllegalArgumentException("Offset table must use
database.table format");
}
+ this.database = tableParts[0];
+ this.table = tableParts[1];
this.insertSql =
"INSERT INTO "
- + quoteIdentifier(tableParts[0])
+ + quoteIdentifier(database)
+ "."
- + quoteIdentifier(tableParts[1])
+ + quoteIdentifier(table)
+ " (`consumer_id`, `offset_timestamp`, `update_time`)
"
+ "VALUES (?, ?, CURRENT_TIMESTAMP(3))";
this.consumerId = consumerId;
}
+ public void validateOffsetTable() {
+ try {
+ Connection connection =
connectionProvider.getOrEstablishConnection();
+ DatabaseMetaData metadata = connection.getMetaData();
+ try (ResultSet tables = metadata.getTables(database, null, table,
null)) {
+ while (tables.next()) {
+ if (table.equals(tables.getString("TABLE_NAME"))) {
+ return;
+ }
+ }
+ }
+ } catch (Exception error) {
+ throw new DorisRuntimeException(
+ "Failed to validate offset table: " + database + "." +
table, error);
+ }
+ throw new DorisRuntimeException("Offset table does not exist: " +
database + "." + table);
+ }
+
/** Invokes the callback with null on success or the publication error on
failure. */
public void publish(String offset, Consumer<Exception> callback) {
Exception failure = null;
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisOffsetPublisherTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisOffsetPublisherTest.java
index d3a1e40d..5ecb0717 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisOffsetPublisherTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisOffsetPublisherTest.java
@@ -21,11 +21,14 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.sql.Connection;
+import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -35,15 +38,55 @@ import static org.mockito.Mockito.when;
class DorisOffsetPublisherTest {
private final JdbcConnectionProvider connectionProvider =
mock(JdbcConnectionProvider.class);
private final Connection connection = mock(Connection.class);
+ private final DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+ private final ResultSet tables = mock(ResultSet.class);
private final PreparedStatement statement = mock(PreparedStatement.class);
@BeforeEach
void setUp() throws Exception {
when(connectionProvider.getOrEstablishConnection()).thenReturn(connection);
+ when(connection.getMetaData()).thenReturn(metadata);
+ when(metadata.getTables("ops", null, "flink_source_offsets",
null)).thenReturn(tables);
when(connection.prepareStatement(any())).thenReturn(statement);
when(statement.executeUpdate()).thenReturn(1);
}
+ @Test
+ void validatesExistingOffsetTable() throws Exception {
+ when(tables.next()).thenReturn(true);
+
when(tables.getString("TABLE_NAME")).thenReturn("flink_source_offsets");
+ DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ connectionProvider, "ops.flink_source_offsets",
"prod.sales.orders");
+
+
assertThatCode(publisher::validateOffsetTable).doesNotThrowAnyException();
+ }
+
+ @Test
+ void rejectsWildcardTableNameMatch() throws Exception {
+ when(tables.next()).thenReturn(true, false);
+
when(tables.getString("TABLE_NAME")).thenReturn("flinkXsourceXoffsets");
+ DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ connectionProvider, "ops.flink_source_offsets",
"prod.sales.orders");
+
+ assertThatThrownBy(publisher::validateOffsetTable)
+ .hasMessageContaining("Offset table does not exist:
ops.flink_source_offsets");
+ }
+
+ @Test
+ void failsValidationWhenJdbcDriverIsMissing() throws Exception {
+ when(connectionProvider.getOrEstablishConnection())
+ .thenThrow(new
ClassNotFoundException("com.mysql.cj.jdbc.Driver"));
+ DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ connectionProvider, "ops.flink_source_offsets",
"prod.sales.orders");
+
+ assertThatThrownBy(publisher::validateOffsetTable)
+ .hasMessageContaining("Failed to validate offset table:
ops.flink_source_offsets")
+ .hasCauseInstanceOf(ClassNotFoundException.class);
+ }
+
@Test
void writesCompletedOffsetWithPreparedStatement() throws Exception {
DorisOffsetPublisher publisher =
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
index 7652c1a6..1ce5ef22 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
@@ -33,11 +33,13 @@ import org.apache.flink.util.Preconditions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
import org.apache.doris.flink.deserialization.DorisDeserializationSchema;
import org.apache.doris.flink.source.assigners.DorisSourceSplitAssigner;
import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
import
org.apache.doris.flink.source.enumerator.DorisSourceCheckpointSerializer;
import org.apache.doris.flink.source.enumerator.DorisSourceEnumerator;
+import org.apache.doris.flink.source.reader.DorisOffsetPublisher;
import org.apache.doris.flink.source.reader.DorisRecordEmitter;
import org.apache.doris.flink.source.reader.DorisSourceFetcherManager;
import org.apache.doris.flink.source.reader.DorisSourceReader;
@@ -240,6 +242,13 @@ public class DorisSource<OUT>
Preconditions.checkNotNull(
options.getJdbcUrl(),
"jdbc-url is required when source.binlog.offset-table
is configured");
+ try (DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ new SimpleJdbcConnectionProvider(options),
+ offsetTable,
+ consumerId)) {
+ publisher.validateOffsetTable();
+ }
}
return new DorisSource<>(options, readOptions, boundedness,
deserializer);
}
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
index d40273ee..5c9eb34b 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
@@ -23,11 +23,18 @@ import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import java.sql.DriverManager;
+import java.sql.SQLException;
import java.util.List;
+import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mockStatic;
class DorisSourceOptionsTest {
@@ -190,16 +197,31 @@ class DorisSourceOptionsTest {
.setBinlogConsumerId("prod.sales.orders")
.build()))
.hasMessageContaining("jdbc-url");
- assertThat(
- buildSource(
- DorisReadOptions.builder()
-
.setScanMode(DorisSourceScanMode.LATEST)
-
.setBinlogOffsetTable("ops.flink_source_offsets")
-
.setBinlogConsumerId("prod.sales.orders")
- .build(),
- "db.table",
- "jdbc:mysql://127.0.0.1:9030"))
- .isNotNull();
+ }
+
+ @Test
+ void validatesOffsetTableBeforeCreatingSource() {
+ SQLException failure = new SQLException("unavailable");
+ try (MockedStatic<DriverManager> driverManager =
mockStatic(DriverManager.class)) {
+ driverManager
+ .when(() -> DriverManager.getConnection(anyString(),
any(Properties.class)))
+ .thenThrow(failure);
+
+ assertThatThrownBy(
+ () ->
+ buildSource(
+ DorisReadOptions.builder()
+
.setScanMode(DorisSourceScanMode.LATEST)
+ .setBinlogOffsetTable(
+
"ops.flink_source_offsets")
+
.setBinlogConsumerId("prod.sales.orders")
+ .build(),
+ "db.table",
+ "jdbc:mysql://127.0.0.1:9030"))
+ .hasMessageContaining(
+ "Failed to validate offset table:
ops.flink_source_offsets")
+ .hasCauseInstanceOf(SQLException.class);
+ }
}
@Test
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
index 8edb6ef6..6a323a57 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
@@ -31,11 +31,13 @@ import org.apache.flink.util.Preconditions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
import org.apache.doris.flink.deserialization.DorisDeserializationSchema;
import org.apache.doris.flink.source.assigners.DorisSourceSplitAssigner;
import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
import
org.apache.doris.flink.source.enumerator.DorisSourceCheckpointSerializer;
import org.apache.doris.flink.source.enumerator.DorisSourceEnumerator;
+import org.apache.doris.flink.source.reader.DorisOffsetPublisher;
import org.apache.doris.flink.source.reader.DorisRecordEmitter;
import org.apache.doris.flink.source.reader.DorisSourceFetcherManager;
import org.apache.doris.flink.source.reader.DorisSourceReader;
@@ -235,6 +237,13 @@ public class DorisSource<OUT>
Preconditions.checkNotNull(
options.getJdbcUrl(),
"jdbc-url is required when source.binlog.offset-table
is configured");
+ try (DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ new SimpleJdbcConnectionProvider(options),
+ offsetTable,
+ consumerId)) {
+ publisher.validateOffsetTable();
+ }
}
return new DorisSource<>(options, readOptions, boundedness,
deserializer);
}
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
index f3374f31..de9e7130 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
@@ -43,7 +43,6 @@ import org.apache.doris.flink.sink.batch.DorisBatchSink;
import org.apache.doris.flink.sink.writer.serializer.SimpleStringSerializer;
import org.apache.doris.flink.table.DorisConfigOptions;
import org.apache.doris.flink.utils.MockSource;
-import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -51,6 +50,7 @@ import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.sql.Connection;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
@@ -639,10 +639,7 @@ public class DorisSinkITCase extends AbstractITCaseService
{
String query =
String.format("select id,task_id from %s.%s order by 1,2",
DATABASE, TABLE_CSV_JM);
- List<String> actualResult =
- ContainerUtils.getResult(getDorisQueryConnection(), LOG,
expected, query, 2);
- Assert.assertTrue(
- actualResult.size() >= expected.size() &&
actualResult.containsAll(expected));
+ waitForExpectedResult(expected, query);
}
@Test
@@ -706,10 +703,22 @@ public class DorisSinkITCase extends
AbstractITCaseService {
String query =
String.format("select id,task_id from %s.%s order by 1,2",
DATABASE, TABLE_CSV_TM);
- List<String> actualResult =
- ContainerUtils.getResult(getDorisQueryConnection(), LOG,
expected, query, 2);
- Assert.assertTrue(
- actualResult.size() >= expected.size() &&
actualResult.containsAll(expected));
+ waitForExpectedResult(expected, query);
+ }
+
+ private void waitForExpectedResult(List<String> expected, String query)
throws Exception {
+ try (Connection connection = getDorisQueryConnection()) {
+ waitUntilCondition(
+ () -> {
+ List<String> actualResult =
+ ContainerUtils.getResult(connection, LOG,
expected, query, 2);
+ return actualResult.size() >= expected.size()
+ && actualResult.containsAll(expected);
+ },
+ Deadline.fromNow(Duration.ofSeconds(30)),
+ 200,
+ "Sink result did not become visible in time.");
+ }
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
