github-actions[bot] commented on code in PR #68270:
URL: https://github.com/apache/doris/pull/68270#discussion_r4081645917
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1095,22 +1100,25 @@ private void modifyPropertiesInternal(Map<String,
String> inputProperties) throw
StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
+ if (Config.isCloudMode()) {
+ resetCloudProgress(offset);
Review Comment:
[P1] Keep combined SQL+offset ALTER atomic. The grammar allows PROPERTIES
together with a new DML statement, but by the time this cloud reset runs
alterJob has already installed the new executeSql/baseCommand/originTvfProps. A
non-OK response or RpcException returns before the provider/properties change
and before JobManager writes the edit log, so the failed ALTER leaves only the
current master using the new SQL with the old offset; failover restores the old
SQL. Please stage the SQL-derived state until the fallible reset succeeds (or
restore it on failure), and cover a combined ALTER whose reset fails.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -526,13 +526,46 @@ public Offset deserializeOffsetProperty(String offset) {
}
@Override
- public void validateAlterOffset(String offset) throws Exception {
- if (!DataSourceConfigValidator.isJsonOffset(offset)) {
+ public void validateAlterOffset(String offset) throws AnalysisException {
+ JsonNode offsetNode;
+ try {
+ offsetNode = objectMapper.readTree(offset);
+ } catch (Exception e) {
+ offsetNode = null;
+ }
+ if (offsetNode == null || !offsetNode.isObject()) {
throw new AnalysisException(
"ALTER JOB for CDC only supports JSON specific offset, "
+ "e.g. '{\"file\":\"binlog.000001\",\"pos\":\"154\"}' for
MySQL "
+ "or '{\"lsn\":\"12345678\"}' for PostgreSQL");
}
+
+ boolean valid = switch (sourceType) {
+ case POSTGRES -> isNonNegativeLong(offsetNode.get("lsn"));
+ case MYSQL, OCEANBASE -> offsetNode.has("file") &&
offsetNode.has("pos")
+ ? hasText(offsetNode.path("file")) &&
isNonNegativeLong(offsetNode.get("pos"))
+ : hasText(offsetNode.path("gtids"));
+ default -> throw new AnalysisException(
Review Comment:
[P1] Validate the complete file/position offset with the same contract the
CDC reader consumes. For example,
{"file":"mysql-bin.000001","pos":"154","kind":"BOGUS"} passes this arm because
only file and pos are checked. MySqlSourceReader retains the full map and calls
getOffsetKind() while selecting the next split start, so ALTER succeeds and is
journaled, then the resumed task fails on the invalid enum value. Please reject
invalid retained fields or share the source-specific downstream parser, with a
negative test for an invalid kind.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java:
##########
@@ -560,7 +560,7 @@ protected void validateStreamSource(
+ " committed position without data
loss.",
baseReq.getJobId(), dialect.getSlotName()));
}
- Lsn requestedLsn = extractRequestedLsn(offsetMeta);
+ Lsn requestedLsn = ((PostgresOffset) startingOffset).getLsn();
Lsn restartLsn = slotState.slotRestartLsn();
// restart_lsn must stay <= committed position; a higher one means
the slot was
// recreated
Review Comment:
[P0] Compare against confirmed_flush_lsn, not only restart_lsn. restart_lsn
is the oldest WAL the slot may still need and can remain behind both positions;
the same-name slot can therefore satisfy restart_lsn <= requested while
confirmed_flush_lsn > requested. PostgreSQL starts logical replication at the
greater of the requested LSN and confirmed_flush_lsn, so this guard passes and
silently skips the entire gap. Use slotLastFlushedLsn() as the consumed
boundary (and keep any restart_lsn check needed for older-server failure
classification), then make the regression assert the restart_lsn <= checkpoint
< confirmed_flush_lsn case. See
https://www.postgresql.org/docs/17/protocol-replication.html
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -526,13 +526,46 @@ public Offset deserializeOffsetProperty(String offset) {
}
@Override
- public void validateAlterOffset(String offset) throws Exception {
- if (!DataSourceConfigValidator.isJsonOffset(offset)) {
+ public void validateAlterOffset(String offset) throws AnalysisException {
+ JsonNode offsetNode;
+ try {
+ offsetNode = objectMapper.readTree(offset);
+ } catch (Exception e) {
+ offsetNode = null;
+ }
+ if (offsetNode == null || !offsetNode.isObject()) {
throw new AnalysisException(
"ALTER JOB for CDC only supports JSON specific offset, "
+ "e.g. '{\"file\":\"binlog.000001\",\"pos\":\"154\"}' for
MySQL "
+ "or '{\"lsn\":\"12345678\"}' for PostgreSQL");
}
+
+ boolean valid = switch (sourceType) {
+ case POSTGRES -> isNonNegativeLong(offsetNode.get("lsn"));
+ case MYSQL, OCEANBASE -> offsetNode.has("file") &&
offsetNode.has("pos")
+ ? hasText(offsetNode.path("file")) &&
isNonNegativeLong(offsetNode.get("pos"))
+ : hasText(offsetNode.path("gtids"));
+ default -> throw new AnalysisException(
+ "Unsupported CDC source type for ALTER JOB offset: " +
sourceType);
+ };
Review Comment:
[P1] Make the accepted GTID-only form reach the split reader. This branch
accepts a bare {"gtids":...} map and FE persists/sends it, but
MySqlSourceReader.createBinlogSplit only injects kind=SPECIFIC when a file key
exists. The bare map consequently has null kind/file and the reader falls back
to the unchanged startup offset (or earliest), silently ignoring the ALTER;
meanwhile the normalized ofGtidSet map with kind plus empty file/pos
placeholders is rejected by the file/position arm here. Normalize and validate
one representation consistently in both paths, and add an end-to-end
MySQL/OceanBase ALTER-resume test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -526,13 +526,46 @@ public Offset deserializeOffsetProperty(String offset) {
}
@Override
- public void validateAlterOffset(String offset) throws Exception {
- if (!DataSourceConfigValidator.isJsonOffset(offset)) {
+ public void validateAlterOffset(String offset) throws AnalysisException {
+ JsonNode offsetNode;
+ try {
+ offsetNode = objectMapper.readTree(offset);
+ } catch (Exception e) {
+ offsetNode = null;
+ }
+ if (offsetNode == null || !offsetNode.isObject()) {
throw new AnalysisException(
"ALTER JOB for CDC only supports JSON specific offset, "
+ "e.g. '{\"file\":\"binlog.000001\",\"pos\":\"154\"}' for
MySQL "
+ "or '{\"lsn\":\"12345678\"}' for PostgreSQL");
}
+
+ boolean valid = switch (sourceType) {
+ case POSTGRES -> isNonNegativeLong(offsetNode.get("lsn"));
+ case MYSQL, OCEANBASE -> offsetNode.has("file") &&
offsetNode.has("pos")
+ ? hasText(offsetNode.path("file")) &&
isNonNegativeLong(offsetNode.get("pos"))
+ : hasText(offsetNode.path("gtids"));
Review Comment:
[P1] Reject zero for an ALTERed PostgreSQL LSN. Here 0 is accepted, but
PostgresSourceReader treats it as the internal initial sentinel and
deliberately skips the slot-position guard for requestedLsn <= 0. This ALTER
path installs a specific stream offset, so a user-requested rewind to 0 can
instead resume at the slot's later confirmed position without reporting the
skipped range. Keep zero for internal initial startup only and require a
positive LSN here.
--
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]
