yzeng1618 commented on code in PR #10951:
URL: https://github.com/apache/seatunnel/pull/10951#discussion_r3338179218
##########
seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java:
##########
@@ -266,6 +290,155 @@ public void notifyCheckpointComplete(long checkpointId)
throws Exception {
tableId,
eventTime);
+ applyNextPendingSchemaChange();
+ }
+
+ /**
+ * Handles fallback timer firing on the task thread (Flink 1.13 checkpoint
stall workaround).
+ *
+ * <p>IMPORTANT: This method still respects the checkpoint-completion
safety fence. The DDL is
+ * only applied if at least one checkpoint has completed after the schema
event was detected
+ * ({@code firstSeenCheckpointId >= 0}). This ensures XA transactions from
the earlier
+ * checkpoint cycle have finished before ALTER TABLE runs.
+ *
+ * <p>If no checkpoint has completed yet, we reschedule the fallback timer
and wait. The
+ * fallback only kicks in when checkpoints stall AFTER the first post-DDL
checkpoint has
+ * completed, which is the scenario in Flink 1.13 high-parallelism CDC
jobs where some source
+ * subtasks finish before others.
+ */
+ private void handleFallbackTimerOnTaskThread() throws InterruptedException
{
+ fallbackTimerFired = false;
+
+ if (!schemaChangePending || pendingQueue.isEmpty()) {
+ return;
+ }
+
+ if (lastCheckpointCompletedMs > 0
+ && System.currentTimeMillis() - lastCheckpointCompletedMs
+ < CHECKPOINT_STALL_TIMEOUT_MS) {
+ scheduleFallbackTimer();
+ return;
+ }
+
+ BufferedRecord head = advancePastDataRecords();
+ if (head == null) {
+ return;
+ }
+
+ if (firstSeenCheckpointId < 0) {
+ log.info(
+ "Fallback timer fired but no checkpoint has completed
after schema event "
+ + "for table {} (epoch {}). Rescheduling fallback
to preserve "
+ + "checkpoint-completion safety fence.",
+ head.schemaEvent.tableIdentifier(),
+ head.schemaEvent.getCreatedTime());
+ scheduleFallbackTimer();
+ return;
+ }
+
+ log.warn(
+ "Checkpoint stall detected after first post-DDL checkpoint {}.
"
+ + "Applying deferred DDL for table {} (epoch {}) via
fallback timer. "
+ + "Note: data committed via normal Flink checkpoint
lifecycle may be "
+ + "delayed until checkpoints resume.",
+ firstSeenCheckpointId,
+ head.schemaEvent.tableIdentifier(),
+ head.schemaEvent.getCreatedTime());
+
+ applyNextPendingSchemaChange();
+ }
+
+ private void scheduleFallbackTimer() {
+ if (fallbackScheduler == null || fallbackScheduler.isShutdown()) {
+ return;
+ }
+ ScheduledFuture<?> existing = pendingFallbackFuture;
+ if (existing != null && !existing.isDone()) {
+ return;
+ }
+ pendingFallbackFuture =
+ fallbackScheduler.schedule(
+ () -> {
+ fallbackTimerFired = true;
+ registerProcessingTimeCallback();
+ },
+ CHECKPOINT_STALL_TIMEOUT_MS,
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void registerProcessingTimeCallback() {
Review Comment:
registerProcessingTimeCallback() uses reflection throughout to infer the
method names of Flink’s internal APIs. Does this introduce potential risks? For
instance, any change to method names or parameters would break the logic. Could
we instead register the ProcessingTimeCallback via the strongly-typed public
API exposed by getProcessingTimeService()?
##########
seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java:
##########
@@ -116,20 +140,25 @@ public void open() throws Exception {
}
@Override
- public void processElement(StreamRecord<SeaTunnelRow> streamRecord) {
+ public void processElement(StreamRecord<SeaTunnelRow> streamRecord)
+ throws InterruptedException {
SeaTunnelRow element = streamRecord.getValue();
if (!isSchemaEvolutionEnabled(pluginConfig)) {
output.collect(streamRecord);
return;
}
+ if (fallbackTimerFired) {
+ handleFallbackTimerOnTaskThread();
+ }
Review Comment:
One entry of the fallback logic in SchemaOperator relies on continuous
invocations of processElement. However, this PR addresses cases where no more
data flows from the source and processElement is no longer triggered, rendering
this path invalid. Consequently, fallback processing entirely depends on the
timer registered via reflection.
If reflective registration fails, the logic falls back to a flag-based
implementation that also requires processElement to run. When there is no
incoming data, neither fallback path can be executed, the fallback mechanism
fails to work, and the task hanging issue remains unresolved under reflection
failure scenarios.
##########
seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java:
##########
@@ -218,20 +227,26 @@ public enum SchemaProcessingStatus {
public boolean requestSchemaChange(TableIdentifier tableId, long epoch,
long timeoutMs)
throws InterruptedException, SchemaCoordinationException {
String key = tableId.toString() + "#" + epoch;
- int expectedAcks = activeSinkSubtasks.size();
- if (expectedAcks == 0) {
+ int totalSubtasks = activeSinkSubtasks.size();
+ if (totalSubtasks == 0) {
log.warn(
"No active sink subtasks. Cannot coordinate schema change
for table {} (epoch {}). "
+ "Assuming success to avoid deadlock.",
tableId,
epoch);
return true;
}
+ // Schema changes (DDL) are database-level operations that only need
to execute once.
+ // Due to Flink's partitioning, only one subtask receives the schema
change event,
+ // so we only need 1 ACK to confirm the DDL was applied successfully.
+ int expectedAcks = 1;
Review Comment:
Completing the request on a single ACK is correct for "the DDL runs once on
the DB", but it implicitly assumes every sink subtask's local schema view is
refreshed by the broadcast/State path rather than by receiving the event. Could
we add a comment stating this precondition, and a multi-table (table-names with
≥2 tables) regression test so the "view refresh covers all subtasks" invariant
is guarded? Current E2E only covers single-table.
--
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]
