morningman opened a new pull request, #710:
URL: https://github.com/apache/doris-flink-connector/pull/710
# Proposed changes
Issue Number: None (follows apache/doris#69311)
## Problem Summary:
Doris renamed the experimental row binlog feature to **changelog** in
apache/doris#69311 (merged into Doris master as `73a987e6059`). The incremental
source added in #679 reads that feature, and two kinds of names it depends on
changed:
- A table turns the feature on with `"changelog.enable" = "true"` instead of
`"binlog.enable" = "true"` plus `"binlog.format" = "ROW"`.
`binlog.need_historical_value` became `changelog.need_historical_value`, and
the row-binlog use of `binlog.ttl_seconds` became `changelog.ttl_seconds`.
Doris now rejects `binlog.format` and `binlog.need_historical_value` in every
DDL.
- The hidden columns that the source selects and orders by are now
`__DORIS_CHANGELOG_TSO__`, `__DORIS_CHANGELOG_LSN__` and
`__DORIS_CHANGELOG_OP__` (were `__DORIS_BINLOG_*__`).
The `binlog()` table function also became `changelog()`, but the connector
does not use it. The `@incr(...)` syntax, the `DETAIL` / `MIN_DELTA` /
`APPEND_ONLY` increment types, the op values, `information_schema.tso_status`
and the visible-wait error code `5101` that the source retries on did not
change.
The incremental source in connector 26.2.0 and 26.3.0 builds `SELECT ...,
__DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__, __DORIS_BINLOG_OP__ FROM
db.t@incr(...) ORDER BY ...` and looks for those column names in the Arrow
result. It therefore only works with Doris master builds from before the
rename. Against Doris master after the rename, that query fails on a changelog
table, and the table DDL used by the integration fixtures is rejected:
```
-- the incremental query built by the current connector
ERROR 1105 (HY000): errCode = 2, detailMessage = Unknown column
'__DORIS_BINLOG_TSO__' in 'table list' in PROJECT clause(line 1, pos 69)
-- CREATE TABLE ... ("binlog.enable" = "true", "binlog.format" = "ROW",
"binlog.need_historical_value" = "true")
ERROR 1105 (HY000): errCode = 2, detailMessage = Property binlog.format is
no longer supported, use "changelog.enable" = "true" to enable changelog, or
remove it for CCR binlog
```
The same query with `__DORIS_CHANGELOG_*__` returns the expected rows (op 0
/ 2 / 3 / 1 for insert / update-before / update-after / delete, which is what
`DorisSourceRecord` maps to `RowKind`).
### What this PR changes
1. The incremental query and the Arrow reader use the new hidden column
names.
2. Every row-binlog name in the connector is renamed to changelog, the Doris
feature name: option keys, constants, the increment-type enum,
`DorisReadOptions` fields, getters and builder setters, the `DorisSourceRecord`
metadata getters, validation and exception messages, and javadoc. The option
keys keep the existing `source.` prefix and their suffixes.
3. The integration-test and E2E fixtures create their tables with
`"changelog.enable" = "true"` and `"changelog.need_historical_value" = "true"`.
Not changed: MySQL binlog / Flink CDC / Debezium code and test data (those
are MySQL's binlog), Flink's own `ChangelogMode` / `RowKind` handling, and the
split and enumerator checkpoint formats (they store split ids and timestamps
only, no renamed names).
| Kind | Old | New |
|---|---|---|
| Option key | `source.binlog.increment-type` |
`source.changelog.increment-type` |
| Option key | `source.binlog.poll-interval` |
`source.changelog.poll-interval` |
| Option key | `source.binlog.visible-wait-timeout` |
`source.changelog.visible-wait-timeout` |
| Option key | `source.binlog.offset-table` |
`source.changelog.offset-table` |
| Option key | `source.binlog.consumer-id` | `source.changelog.consumer-id` |
| `DorisConfigOptions` | `SOURCE_BINLOG_INCREMENT_TYPE`,
`SOURCE_BINLOG_POLL_INTERVAL`, `SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT`,
`SOURCE_BINLOG_OFFSET_TABLE`, `SOURCE_BINLOG_CONSUMER_ID` |
`SOURCE_CHANGELOG_*` with the same suffixes |
| `ConfigurationOptions` | `SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT` |
`SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT` |
| Class | `org.apache.doris.flink.source.DorisBinlogIncrementType` |
`org.apache.doris.flink.source.DorisChangelogIncrementType` |
| `DorisReadOptions` | `getBinlogIncrementType()`,
`getBinlogPollIntervalMs()`, `getBinlogVisibleWaitTimeoutMs()`,
`getBinlogOffsetTable()`, `getBinlogConsumerId()` and the matching
`Builder.setBinlog*()` | `getChangelog*()` / `Builder.setChangelog*()` |
| `DorisSourceRecord` (`@PublicEvolving`) | `getBinlogTso()`,
`getBinlogLsn()`, `getBinlogOp()` | `getChangelogTso()`, `getChangelogLsn()`,
`getChangelogOp()` |
| Hidden columns read by the source | `__DORIS_BINLOG_TSO__`,
`__DORIS_BINLOG_LSN__`, `__DORIS_BINLOG_OP__` | `__DORIS_CHANGELOG_TSO__`,
`__DORIS_CHANGELOG_LSN__`, `__DORIS_CHANGELOG_OP__` |
| Doris table properties in the IT / E2E fixtures | `"binlog.enable" =
"true"` + `"binlog.format" = "ROW"`, `"binlog.need_historical_value"` |
`"changelog.enable" = "true"`, `"changelog.need_historical_value"` |
### Compatibility
- This is a hard rename with no deprecated keys. A Flink table that still
sets a `source.binlog.*` option fails the factory's option validation as an
unsupported option, and Java code that uses the renamed class or methods has to
be updated (for example a custom deserialization schema that calls
`DorisSourceRecord.getBinlogTso()`).
- The old keys shipped in 26.2.0 (four of them; `visible-wait-timeout` came
in 26.3.0) and 26.3.0, but they only ever worked against unreleased Doris
master builds from before the rename. A connector with this change cannot read
those builds either, because the hidden column names differ, and Doris does not
read tables created with `"binlog.format" = "ROW"` after the rename: they have
to be dropped and recreated with `"changelog.enable" = "true"`. Users of the
incremental source therefore move Doris, their tables and the connector
together. If maintainers prefer to keep old Flink DDL parsing, the five options
can take `withDeprecatedKeys(...)`; I left that out on purpose and can add it.
- Doris has no release with changelog yet. A connector release with this
change is needed before, or together with, the first Doris release that
contains it; until then no released connector can read changelog tables.
## Checklist(Required)
1. Does it affect the original behavior: Yes. The five `source.binlog.*`
option keys and the listed class and method names are renamed, and the
incremental source reads the renamed Doris hidden columns.
2. Has unit tests been added: No Need. The existing unit tests, integration
tests and the incremental E2E case are updated to the new names and pass (see
below).
3. Has document been added or modified: No. Nothing in this repository
documents these options; the Flink connector documentation on doris.apache.org
lives outside this repo and needs the same rename there if it describes the
incremental source.
4. Does it need to update dependencies: No
5. Are there any changes that cannot be rolled back: No
## Further comments
Tests, all on JDK 17 (CI builds Flink 1.x with JDK 8; only JDK 17 was
available locally):
- Unit tests, `mvn clean test -Pflink1 -pl flink-doris-connector-flink1 -am`
and `mvn clean test -Pflink2 -pl flink-doris-connector-flink2 -am`. Spotless
check passes as part of both builds.
| Profile | Module | Tests run | Failures | Errors | Skipped |
|---|---|---|---|---|---|
| flink1 (Flink 1.20.0) | base | 455 | 0 | 0 | 1 |
| flink1 (Flink 1.20.0) | flink1 | 67 | 1 | 0 | 1 |
| flink2 (Flink 2.3.0) | base | 455 | 0 | 0 | 1 |
| flink2 (Flink 2.3.0) | flink2 | 20 | 0 | 0 | 0 |
The counts are the same as on master `1050cd0` on the same machine. The
one failure, `DatabaseSyncTest.getJdbcUrlTemplateTest`, fails identically on
master and is unrelated: `Db2DatabaseSync.getJdbcUrlTemplate` iterates a
`java.util.Properties`, whose iteration order changed after JDK 8, and the test
expects the JDK 8 order.
- Incremental integration tests against a single-FE / single-BE Doris master
build that contains apache/doris#69311. These cases only run in customer-env
mode, so the CI workflows (which use the 4.1.3 Docker image) skip them:
```
mvn clean test -Pflink2 -pl flink-doris-connector-it -am \
-DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false \
-Dtest=DorisIncrementalSourceITCase,DorisIncrementalSourceFailoverITCase,DorisIncrementalSourceE2ECase
\
-Dcustomer_env=true -Ddoris_host=127.0.0.1 -Ddoris_query_port=<fe query
port> \
-Ddoris_http_port=<fe http port> -Ddoris_user=root -Ddoris_passwd= \
-Ddoris_flight_sql_port=<fe arrow flight sql port>
```
| Profile | `DorisIncrementalSourceITCase` |
`DorisIncrementalSourceFailoverITCase` (JM, TM) |
`DorisIncrementalSourceE2ECase` |
|---|---|---|---|
| flink2 (Flink 2.3.0) | 6 / 6 pass | 2 / 2 pass | 1 / 1 pass |
| flink1 (Flink 1.20.0) | 6 / 6 pass | 2 / 2 pass | 1 / 1 pass |
- After this change, the only remaining `binlog` strings in the repository
are MySQL's: Debezium source metadata in test JSON (`mysql_binlog_source`,
MySQL binlog file names), the Flink CDC MySQL `BinlogOffset` API in
`MysqlDatabaseSync`, and the job name of a MySQL CDC example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
