CloverDew commented on code in PR #10951:
URL: https://github.com/apache/seatunnel/pull/10951#discussion_r3338476993
##########
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:
The multi-table E2E test itself is a follow-up item and has been noted in
the comment for tracking.
##########
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:
okay, thank you for reviewing. I plan to remove
`registerProcessingTimeCallback()` and all its reflection code.
SchemaOperator13 overrides the `scheduleFallbackTimer()` method, and
`ProcessingTimeService` and `ProcessingTimeCallback` are directly imported as
compile-time types into the `seatunnel-translation-flink-13` module, so they
will not be broken by method renaming in future Flink versions.
##########
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:
okay, I plan to remove the `fallbackTimerFired` flag and the `if` check in
`processElement`. `handleFallbackTimerOnTaskThread()` will now be called
directly from the `ProcessingTimeService` timer callback function, so it will
fire on the Flink task thread regardless of whether source data arrives. This
avoids the blocking fallback path of "reflection failure -> setting flags ->
waiting for `processElement`".
--
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]
