You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by "lvyanquan (via GitHub)" <gi...@apache.org> on 2024/03/12 07:01:21 UTC
[PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
lvyanquan opened a new pull request, #3128:
URL: https://github.com/apache/flink-cdc/pull/3128
This closed https://github.com/apache/flink-cdc/issues/2910#issuecomment-1987707544.
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1554559194
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,4 +136,13 @@ RESPONSE sendRequestToCoordinator(REQUEST request) {
"Failed to send request to coordinator: " + request.toString(), e);
}
}
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
+ if (getRuntimeContext().getIndexOfThisSubtask() == 0) {
Review Comment:
Because there may be multiple subtasks for the SchemaOperator, but we just want to clear the information in the SchemaRegistryHandler once during restart.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1554559016
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -111,8 +115,13 @@ private SchemaChangeResponse requestSchemaChange(
return sendRequestToCoordinator(new SchemaChangeRequest(tableId, schemaChangeEvent));
}
- private ReleaseUpstreamResponse requestReleaseUpstream() {
- return sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ private void requestReleaseUpstream() throws InterruptedException {
+ CoordinationResponse coordinationResponse =
+ sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ while (coordinationResponse instanceof SchemaChangeProcessingResponse) {
Review Comment:
Done, add a new pipeline config https://github.com/apache/flink-cdc/pull/3128/files#diff-efc0e20a0c40e81813d3be00eadab7bc066598e521b24a85f2c3e5c9b83c2b4bR92 to control this.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/event/SchemaChangeResultRequest.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.schema.event;
+
+import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
+
+/** request for get change result. */
Review Comment:
done.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "fanqiejiang8 (via GitHub)" <gi...@apache.org>.
fanqiejiang8 commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2002855730
@lvyanquan @BaoPiao When task parallelism is multiple, task retry all sent SchemaRegistry task more the RefreshPendingListsRequest request, that each request will not cause data inconsistency pendingSchemaChanges
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "hk-lrzy (via GitHub)" <gi...@apache.org>.
hk-lrzy commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1609878354
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,11 +142,23 @@ public Schema load(TableId tableId) {
});
}
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
Review Comment:
why only restore need refresh.
does there have follow case:
1. job start
2. schema operator received some schema change events and send it to coordinator.
3. operator failed before schema operator get response and there have no checkpoint.
4. schema operator restart and send new `CreateTableEvent` to the coordinator.
5. coordinator have some pending event and cause new `CreateTableEvent` will in the pending list, and schema operator will blocking forever.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1524312973
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -70,6 +81,8 @@ public SchemaRegistryRequestHandler(
this.flushedSinkWriters = new HashSet<>();
this.pendingSchemaChanges = new LinkedList<>();
this.schemaManager = schemaManager;
+ schemaChangeThreadPool = Executors.newSingleThreadExecutor();
+ isSchemaChangeApplying = true;
Review Comment:
done.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-1994021777
Thanks @ruanhang1993 for these suggestion, address it.
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "hk-lrzy (via GitHub)" <gi...@apache.org>.
hk-lrzy commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1609878354
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,11 +142,23 @@ public Schema load(TableId tableId) {
});
}
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
Review Comment:
why only restored from checkpoint need refresh request?
does there have follow case:
1. job start
2. schema operator received some schema change events and send it to coordinator.
3. operator failed before schema operator get response and there have no checkpoint.
4. schema operator restart and send new `CreateTableEvent` to the coordinator.
5. coordinator have some pending event and cause new `CreateTableEvent` will in the pending list, and schema operator will blocking forever.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1549367497
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -111,8 +115,13 @@ private SchemaChangeResponse requestSchemaChange(
return sendRequestToCoordinator(new SchemaChangeRequest(tableId, schemaChangeEvent));
}
- private ReleaseUpstreamResponse requestReleaseUpstream() {
- return sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ private void requestReleaseUpstream() throws InterruptedException {
+ CoordinationResponse coordinationResponse =
+ sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ while (coordinationResponse instanceof SchemaChangeProcessingResponse) {
Review Comment:
Continuous loop maybe not acceptable as this may lead endless coordination. Could we introduce a timeout config option with a reasonable default value for coordination ?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/event/SchemaChangeResultRequest.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.schema.event;
+
+import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
+
+/** request for get change result. */
Review Comment:
Could you improve your java doc refer other classes?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,4 +136,13 @@ RESPONSE sendRequestToCoordinator(REQUEST request) {
"Failed to send request to coordinator: " + request.toString(), e);
}
}
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
+ if (getRuntimeContext().getIndexOfThisSubtask() == 0) {
Review Comment:
I didn't catch up this limitation, could you explain this ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2041034979
> Thanks @lvyanquan for the fix, I left some comments. And I think tests is necessary, e.g. a test mocked RPC timeout when processing schema change event is welcome.
Add new tests in https://github.com/apache/flink-cdc/pull/3128/files#diff-c3455d69be8d51aab2230d40c29944f6d451116539098421169a38a7469ed3c7R114.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1521158673
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -62,6 +67,12 @@ public class SchemaRegistryRequestHandler {
private final List<PendingSchemaChange> pendingSchemaChanges;
/** Sink writers which have sent flush success events for the request. */
private final Set<Integer> flushedSinkWriters;
+ /** Status of the execution of current schema change request. */
+ private boolean succeedApplyChange;
+ /** Actual exception if failed to apply schema change. */
+ private Exception actualException;
+ /** Executor service to execute schema change. */
+ private final ExecutorService threadPool;
Review Comment:
threadPool -> schemaChangeThreadPool
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -80,7 +93,13 @@ public SchemaRegistryRequestHandler(
*/
private void applySchemaChange(TableId tableId, SchemaChangeEvent changeEvent) {
LOG.debug("Apply schema change {} to table {}.", changeEvent, tableId);
- metadataApplier.applySchemaChange(changeEvent);
+ succeedApplyChange = false;
Review Comment:
isSchemaChangeApplying = true;
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/event/WaitChangeResultResponse.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.schema.event;
+
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.runtime.operators.schema.SchemaOperator;
+import org.apache.flink.cdc.runtime.operators.schema.coordinator.SchemaRegistry;
+import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
+
+/**
+ * The response for {@link GetChangeResultRequest} or {@link ReleaseUpstreamRequest} from {@link
+ * SchemaRegistry} to {@link SchemaOperator} if not apply {@link SchemaChangeEvent} in time.
+ */
+public class WaitChangeResultResponse implements CoordinationResponse {
Review Comment:
WaitChangeResultResponse -> SchemaChangeProcessingResponse
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -80,7 +93,13 @@ public SchemaRegistryRequestHandler(
*/
private void applySchemaChange(TableId tableId, SchemaChangeEvent changeEvent) {
LOG.debug("Apply schema change {} to table {}.", changeEvent, tableId);
- metadataApplier.applySchemaChange(changeEvent);
+ succeedApplyChange = false;
+ try {
+ metadataApplier.applySchemaChange(changeEvent);
+ succeedApplyChange = true;
+ } catch (Exception e) {
+ actualException = e;
+ }
Review Comment:
try {
metadataApplier.applySchemaChange(changeEvent);
} catch (Exception e) {
this.schemaChangeException = e;
} finally {
this.isSchemaChangeApplying = false;
}
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -140,19 +161,58 @@ public void registerSinkWriter(int sinkSubtask) {
* @param tableId the subtask in SchemaOperator and table that the FlushEvent is about
* @param sinkSubtask the sink subtask succeed flushing
*/
- public void flushSuccess(TableId tableId, int sinkSubtask) {
+ public void flushSuccess(TableId tableId, int sinkSubtask) throws InterruptedException {
flushedSinkWriters.add(sinkSubtask);
if (flushedSinkWriters.equals(activeSinkWriters)) {
LOG.info(
"All sink subtask have flushed for table {}. Start to apply schema change.",
tableId.toString());
PendingSchemaChange waitFlushSuccess = pendingSchemaChanges.get(0);
- applySchemaChange(tableId, waitFlushSuccess.getChangeRequest().getSchemaChangeEvent());
- waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));
+ SchemaChangeEvent schemaChangeEvent =
+ waitFlushSuccess.getChangeRequest().getSchemaChangeEvent();
+ threadPool.submit(
+ () ->
+ applySchemaChange(
+ tableId,
+ waitFlushSuccess.getChangeRequest().getSchemaChangeEvent()));
+ Thread.sleep(1000);
+ if (succeedApplyChange) {
+ if (!(schemaChangeEvent instanceof CreateTableEvent)
+ || !schemaManager.schemaExists(tableId)) {
+ schemaManager.applySchemaChange(schemaChangeEvent);
+ }
+ waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));
+
+ if (RECEIVED_RELEASE_REQUEST.equals(waitFlushSuccess.getStatus())) {
+ startNextSchemaChangeRequest();
+ }
+ } else {
+ if (actualException != null) {
+ throw new RuntimeException("failed to apply schema change.", actualException);
+ }
+ waitFlushSuccess.getResponseFuture().complete(wrap(new WaitChangeResultResponse()));
+ }
Review Comment:
This if should be like this.
```java
if (schemaChangeException != null) {
throw new RuntimeException("failed to apply schema change.", schemaChangeException);
}
if (isSchemaChangeApplying) {
waitFlushSuccess.getResponseFuture().complete(wrap(new WaitChangeResultResponse()));
} else {
if (!(schemaChangeEvent instanceof CreateTableEvent)
|| !schemaManager.schemaExists(tableId)) {
schemaManager.applySchemaChange(schemaChangeEvent);
}
waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));
if (RECEIVED_RELEASE_REQUEST.equals(waitFlushSuccess.getStatus())) {
startNextSchemaChangeRequest();
}
}
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -62,6 +67,12 @@ public class SchemaRegistryRequestHandler {
private final List<PendingSchemaChange> pendingSchemaChanges;
/** Sink writers which have sent flush success events for the request. */
private final Set<Integer> flushedSinkWriters;
+ /** Status of the execution of current schema change request. */
+ private boolean succeedApplyChange;
Review Comment:
succeedApplyChange -> isSchemaChangeApplying
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -62,6 +67,12 @@ public class SchemaRegistryRequestHandler {
private final List<PendingSchemaChange> pendingSchemaChanges;
/** Sink writers which have sent flush success events for the request. */
private final Set<Integer> flushedSinkWriters;
+ /** Status of the execution of current schema change request. */
+ private boolean succeedApplyChange;
+ /** Actual exception if failed to apply schema change. */
+ private Exception actualException;
Review Comment:
actualException -> schemaChangeException
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/event/GetChangeResultRequest.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.schema.event;
+
+import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
+
+/** request for get change result. */
+public class GetChangeResultRequest implements CoordinationRequest {
Review Comment:
GetChangeResultRequest -> SchemaChangeResultRequest
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -140,19 +161,58 @@ public void registerSinkWriter(int sinkSubtask) {
* @param tableId the subtask in SchemaOperator and table that the FlushEvent is about
* @param sinkSubtask the sink subtask succeed flushing
*/
- public void flushSuccess(TableId tableId, int sinkSubtask) {
+ public void flushSuccess(TableId tableId, int sinkSubtask) throws InterruptedException {
flushedSinkWriters.add(sinkSubtask);
if (flushedSinkWriters.equals(activeSinkWriters)) {
LOG.info(
"All sink subtask have flushed for table {}. Start to apply schema change.",
tableId.toString());
PendingSchemaChange waitFlushSuccess = pendingSchemaChanges.get(0);
- applySchemaChange(tableId, waitFlushSuccess.getChangeRequest().getSchemaChangeEvent());
- waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));
+ SchemaChangeEvent schemaChangeEvent =
+ waitFlushSuccess.getChangeRequest().getSchemaChangeEvent();
+ threadPool.submit(
+ () ->
+ applySchemaChange(
+ tableId,
+ waitFlushSuccess.getChangeRequest().getSchemaChangeEvent()));
+ Thread.sleep(1000);
+ if (succeedApplyChange) {
+ if (!(schemaChangeEvent instanceof CreateTableEvent)
+ || !schemaManager.schemaExists(tableId)) {
+ schemaManager.applySchemaChange(schemaChangeEvent);
+ }
+ waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));
+
+ if (RECEIVED_RELEASE_REQUEST.equals(waitFlushSuccess.getStatus())) {
+ startNextSchemaChangeRequest();
+ }
+ } else {
+ if (actualException != null) {
+ throw new RuntimeException("failed to apply schema change.", actualException);
+ }
+ waitFlushSuccess.getResponseFuture().complete(wrap(new WaitChangeResultResponse()));
+ }
+ }
+ }
+ public CompletableFuture<CoordinationResponse> getSchemaChangeResult() {
+ PendingSchemaChange waitFlushSuccess = pendingSchemaChanges.get(0);
+ SchemaChangeEvent schemaChangeEvent =
+ waitFlushSuccess.getChangeRequest().getSchemaChangeEvent();
+ if (succeedApplyChange) {
+ if (!(schemaChangeEvent instanceof CreateTableEvent)
+ || !schemaManager.schemaExists(schemaChangeEvent.tableId())) {
+ schemaManager.applySchemaChange(schemaChangeEvent);
+ }
if (RECEIVED_RELEASE_REQUEST.equals(waitFlushSuccess.getStatus())) {
startNextSchemaChangeRequest();
}
+ return CompletableFuture.supplyAsync(() -> wrap(new ReleaseUpstreamResponse()));
+ } else {
+ if (actualException != null) {
+ throw new RuntimeException("failed to apply schema change.", actualException);
+ }
+ return CompletableFuture.supplyAsync(() -> wrap(new WaitChangeResultResponse()));
Review Comment:
This if should also be changed as the previous comment. Check exception first.
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1524276055
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -70,6 +81,8 @@ public SchemaRegistryRequestHandler(
this.flushedSinkWriters = new HashSet<>();
this.pendingSchemaChanges = new LinkedList<>();
this.schemaManager = schemaManager;
+ schemaChangeThreadPool = Executors.newSingleThreadExecutor();
+ isSchemaChangeApplying = true;
Review Comment:
isSchemaChangeApplying = false;
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "yanghuaiGit (via GitHub)" <gi...@apache.org>.
yanghuaiGit commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2003023705
如果schemaRegister的pending里有多个待执行schema事件，isSchemaChangeApplying代表的是正在执行schema变更事件，但是无法得知是哪个schema事件执行，isSchemaChangeApplying无法和客户端的事件绑定，导致schemaoperator可能需要等待所有的schema事件执行完才会转为非阻塞
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "LYanquan (via GitHub)" <gi...@apache.org>.
LYanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2003214710
@fanqiejiang8 Hi, I have added `getRuntimeContext().getIndexOfThisSubtask() == 0` check to make sure that it won't send many times.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1554795627
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -85,5 +89,12 @@ public class PipelineOptions {
.withDescription(
"The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators.");
+ public static final ConfigOption<Duration> PIPELINE_SCHEMA_OPERATOR_RPC_TIMEOUT =
+ ConfigOptions.key("schema-operator-rpc-timeout")
Review Comment:
```suggestion
ConfigOptions.key("schema-operator.rpc-timeout")
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1567483688
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -85,5 +89,12 @@ public class PipelineOptions {
.withDescription(
"The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators.");
+ public static final ConfigOption<Duration> PIPELINE_SCHEMA_OPERATOR_RPC_TIMEOUT =
+ ConfigOptions.key("schema-operator.rpc-timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(SCHEMA_OPERATOR_RPC_TIMEOUT_SECOND_DEFAULT))
+ .withDescription(
+ "The timeout time for SchemaOperator to wait for schema change. the default value is 3 min.");
Review Comment:
```suggestion
"The timeout time for SchemaOperator to wait downstream SchemaChangeEvent applying finished, the he default value is 3 minutes.");
```
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -23,12 +23,16 @@
import org.apache.flink.cdc.common.configuration.description.Description;
import org.apache.flink.cdc.common.configuration.description.ListElement;
+import java.time.Duration;
+
import static org.apache.flink.cdc.common.configuration.description.TextElement.text;
/** Predefined pipeline configuration options. */
@PublicEvolving
public class PipelineOptions {
+ public static final Integer SCHEMA_OPERATOR_RPC_TIMEOUT_SECOND_DEFAULT = 3 * 60;
Review Comment:
```suggestion
public static final Duration DEFAULT_SCHEMA_OPERATOR_RPC_TIMEOUT = Duration.ofMinutes(3);
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "LYanquan (via GitHub)" <gi...@apache.org>.
LYanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2003238562
> Should we clear the flushedSinkWriters as well?
Thanks for this suggestion, address it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "BaoPiao (via GitHub)" <gi...@apache.org>.
BaoPiao commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2003029727
> @BaoPiao Yeah, this is because SchemaRegistryRequestHandler doesn't clear pendingSchemaChanges when SchemaOperator restarted, I've added one commit [7aba6a4](https://github.com/apache/flink-cdc/commit/7aba6a4489486af9fece954f561dca0c9c443006) to fixed it.
Should we clear the flushedSinkWriters as well?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "hk-lrzy (via GitHub)" <gi...@apache.org>.
hk-lrzy commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1609878354
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,11 +142,23 @@ public Schema load(TableId tableId) {
});
}
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
Review Comment:
why only restore need refresh.
does there have follow case:
1. job start
2. schema operator received some schema change events and send it to coordinator.
3. operator failed before schema operator get response and there have no checkpoint.
4. schema operator restart and send new `CreateTableEvent` to the coordinator.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-1990921705
@ruanhang1993 @PatrickRen PTAL.
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "BaoPiao (via GitHub)" <gi...@apache.org>.
BaoPiao commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-1996731322
If the SchemaRegistryRequestHandler is in the midst of executing a scheme change, and the task manager where the SchemaOperator resides undergoes a restart, could this potentially result in the task being stuck indefinitely?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "LYanquan (via GitHub)" <gi...@apache.org>.
LYanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2002529227
@BaoPiao Yeah, this is because SchemaRegistryRequestHandler doesn't clear pendingSchemaChanges when SchemaOperator restarted, I've added one commit https://github.com/apache/flink-cdc/pull/3128/commits/7aba6a4489486af9fece954f561dca0c9c443006 to fixed it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2071458830
@leonardBang Can you help to start a CI workflow?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2072051290
There is an error in the scenario of using route from CI, and I am trying to fix it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1554795438
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -127,4 +136,13 @@ RESPONSE sendRequestToCoordinator(REQUEST request) {
"Failed to send request to coordinator: " + request.toString(), e);
}
}
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ if (context.isRestored()) {
+ if (getRuntimeContext().getIndexOfThisSubtask() == 0) {
Review Comment:
Got it, it will better to add the explanation as code note
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1554820774
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -85,5 +89,12 @@ public class PipelineOptions {
.withDescription(
"The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators.");
+ public static final ConfigOption<Duration> PIPELINE_SCHEMA_OPERATOR_RPC_TIMEOUT =
+ ConfigOptions.key("schema-operator-rpc-timeout")
Review Comment:
Addressed it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1568374817
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -85,5 +89,12 @@ public class PipelineOptions {
.withDescription(
"The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators.");
+ public static final ConfigOption<Duration> PIPELINE_SCHEMA_OPERATOR_RPC_TIMEOUT =
+ ConfigOptions.key("schema-operator.rpc-timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(SCHEMA_OPERATOR_RPC_TIMEOUT_SECOND_DEFAULT))
+ .withDescription(
+ "The timeout time for SchemaOperator to wait for schema change. the default value is 3 min.");
Review Comment:
Addressed it and rebased to master.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "fanqiejiang8 (via GitHub)" <gi...@apache.org>.
fanqiejiang8 commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1523080473
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -111,8 +113,13 @@ private SchemaChangeResponse requestSchemaChange(
return sendRequestToCoordinator(new SchemaChangeRequest(tableId, schemaChangeEvent));
}
- private ReleaseUpstreamResponse requestReleaseUpstream() {
- return sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ private void requestReleaseUpstream() throws InterruptedException {
+ CoordinationResponse coordinationRequest =
Review Comment:
这个变量的命名有点误解 叫coordinationResponse会不会更好一点
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#discussion_r1523094769
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java:
##########
@@ -111,8 +113,13 @@ private SchemaChangeResponse requestSchemaChange(
return sendRequestToCoordinator(new SchemaChangeRequest(tableId, schemaChangeEvent));
}
- private ReleaseUpstreamResponse requestReleaseUpstream() {
- return sendRequestToCoordinator(new ReleaseUpstreamRequest());
+ private void requestReleaseUpstream() throws InterruptedException {
+ CoordinationResponse coordinationRequest =
Review Comment:
addressed it.
--
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: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34648] add waitChangeResultRequest and WaitChangeResultResponse to avoid RPC timeout. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3128:
URL: https://github.com/apache/flink-cdc/pull/3128#issuecomment-2072115980
> There is an error in the scenario of using route from CI, and I am trying to fix it.
Done fix.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org