yuxiqian commented on code in PR #4541:
URL: https://github.com/apache/flink-cdc/pull/4541#discussion_r4011484408
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java:
##########
@@ -155,6 +165,12 @@ public void close() throws Exception {
/** Restore schema registry state from byte array. */
protected abstract void restore(byte[] checkpointData) throws Exception;
+ /** Stops schema-change work from the previous coordinator generation. */
+ protected abstract void quiesceSchemaChangeExecutor() throws Exception;
+
+ /** Recreates transient state after a coordinator reset. */
+ protected abstract void reinitializeTransientState();
Review Comment:
Shall we call these methods `initialize` and `shutdown`? `initialize` may be
invoked at the first time. And we may need to do more cleaning-up work besides
shutting down executors.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java:
##########
@@ -227,10 +252,9 @@ protected void handleUnrecoverableError(String
taskDescription, Throwable t) {
super.handleUnrecoverableError(taskDescription, t);
// For each pending future, release it exceptionally before quitting
- pendingRequests.forEach(
- (index, tuple) -> {
- tuple.f1.completeExceptionally(t);
- });
+ if (pendingRequests != null) {
Review Comment:
Could you please add some IT case in flink-cdc-composer, and some E2e cases
to cover both regular and distributed topology?
--
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]
