You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2024/08/20 08:40:00 UTC
[PR] [FLINK-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
loserwang1024 opened a new pull request, #3558:
URL: https://github.com/apache/flink-cdc/pull/3558
As shown in https://issues.apache.org/jira/browse/FLINK-36094, Current, SchemaRegistryRequestHandler only throw
SchemaEvolveException, which will not handle the others(like network, oom, or else.
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724858805
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java:
##########
@@ -88,7 +88,7 @@ public abstract class PipelineTestEnvironment extends TestLogger {
@Parameterized.Parameters(name = "flinkVersion: {0}")
public static List<String> getFlinkVersion() {
- return Arrays.asList("1.17.2", "1.18.1", "1.19.1", "1.20.0");
+ return Arrays.asList("1.17.2");
Review Comment:
Why delete other flink versions 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#issuecomment-2298425339
@leonardBang @yuxiqian @ruanhang1993 , CC, WDYT?
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723162033
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -437,6 +444,25 @@ private List<SchemaChangeEvent> lenientizeSchemaChangeEvent(SchemaChangeEvent ev
}
}
+ private boolean shouldIgnoreException(Exception exception) {
+
+ // only UnsupportedSchemaChangeEventException maybe ignore(depends on SchemaChangeBehavior)
Review Comment:
done it
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisMetadataApplier.java:
##########
@@ -221,7 +221,7 @@ private void applyAddColumnEvent(AddColumnEvent event) throws SchemaEvolveExcept
tableId.getSchemaName(), tableId.getTableName(), addFieldSchema);
}
} catch (Exception e) {
- throw new SchemaEvolveException(event, e.getMessage(), e);
+ throw new SchemaEvolveException(event, "fail to apply add column event", e);
Review Comment:
done 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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723028372
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisMetadataApplier.java:
##########
@@ -221,7 +221,7 @@ private void applyAddColumnEvent(AddColumnEvent event) throws SchemaEvolveExcept
tableId.getSchemaName(), tableId.getTableName(), addFieldSchema);
}
} catch (Exception e) {
- throw new SchemaEvolveException(event, e.getMessage(), e);
+ throw new SchemaEvolveException(event, "fail to apply add column event", e);
Review Comment:
```suggestion
throw new SchemaEvolveException(event, "Failed to apply add column event", e);
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -437,6 +444,25 @@ private List<SchemaChangeEvent> lenientizeSchemaChangeEvent(SchemaChangeEvent ev
}
}
+ private boolean shouldIgnoreException(Exception exception) {
+
+ // only UnsupportedSchemaChangeEventException maybe ignore(depends on SchemaChangeBehavior)
Review Comment:
Code flow and comments are confusing here. Something like
```java
// Only `UnsupportedSchemaChangeEventException` could be ignored in TRY_EVOLVE mode.
// In IGNORE mode, will never try to apply schema change events
// In EVOLVE mode, such failure will not be tolerated
// In EXCEPTION mode, an exception will be thrown once captured
return (exception instanceof UnsupportedSchemaChangeEventException) &&
(schemaChangeBehavior == TRY_EVOLVE)
```
might be enough.
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/SchemaEvolveTest.java:
##########
@@ -1040,8 +1040,10 @@ tableId, buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
Column.physicalColumn(
"height", DOUBLE, "Height data")))));
Assertions.assertThatThrownBy(() -> processEvent(schemaOperator, addColumnEvents))
+ .cause()
+ .cause()
.isExactlyInstanceOf(RuntimeException.class)
- .hasMessageContaining("Failed to apply schema change");
+ .hasMessageContaining("failed to apply schema change");
Review Comment:
Use `hasRootCauseInstanceOf` and `hasRootCauseMessage`
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724275532
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/SchemaEvolveTest.java:
##########
@@ -1040,6 +1040,8 @@ tableId, buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
Column.physicalColumn(
"height", DOUBLE, "Height data")))));
Assertions.assertThatThrownBy(() -> processEvent(schemaOperator, addColumnEvents))
+ .cause()
+ .cause()
Review Comment:
I have tried it, but the rootCause is akka error(which can not show the real failed reason in test.
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724672879
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistry.java:
##########
@@ -275,6 +277,8 @@ public void resetToCheckpoint(long checkpointId, @Nullable byte[] checkpointData
throw new IOException(
"Unrecognized serialization version " + schemaManagerSerializerVersion);
}
+ } catch (Throwable throwable) {
+ context.failJob(throwable);
Review Comment:
add it now
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#issuecomment-2298641518
> Could you please modify these parts together?
done 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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723121548
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -147,17 +146,27 @@ private void applySchemaChange(
try {
metadataApplier.applySchemaChange(changeEvent);
LOG.debug("Applied schema change {} to table {}.", changeEvent, tableId);
+ schemaManager.applyEvolvedSchemaChange(changeEvent);
finishedSchemaChanges.add(changeEvent);
- } catch (SchemaEvolveException e) {
+ } catch (Exception e) {
Review Comment:
```suggestion
} catch (Throwable t) {
```
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724437471
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistry.java:
##########
@@ -275,6 +277,8 @@ public void resetToCheckpoint(long checkpointId, @Nullable byte[] checkpointData
throw new IOException(
"Unrecognized serialization version " + schemaManagerSerializerVersion);
}
+ } catch (Throwable throwable) {
+ context.failJob(throwable);
Review Comment:
why not re-throw this `throwable` 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723024778
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/SchemaEvolveTest.java:
##########
@@ -1040,8 +1040,10 @@ tableId, buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
Column.physicalColumn(
"height", DOUBLE, "Height data")))));
Assertions.assertThatThrownBy(() -> processEvent(schemaOperator, addColumnEvents))
+ .cause()
+ .cause()
.isExactlyInstanceOf(RuntimeException.class)
- .hasMessageContaining("Failed to apply schema change");
+ .hasMessageContaining("failed to apply schema change");
Review Comment:
~~Use `hasRootCauseInstanceOf` and `hasRootCauseMessage`~~
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724282834
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/SchemaEvolveTest.java:
##########
@@ -1040,6 +1040,8 @@ tableId, buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
Column.physicalColumn(
"height", DOUBLE, "Height data")))));
Assertions.assertThatThrownBy(() -> processEvent(schemaOperator, addColumnEvents))
+ .cause()
+ .cause()
Review Comment:
OK. I get 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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1724874641
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java:
##########
@@ -88,7 +88,7 @@ public abstract class PipelineTestEnvironment extends TestLogger {
@Parameterized.Parameters(name = "flinkVersion: {0}")
public static List<String> getFlinkVersion() {
- return Arrays.asList("1.17.2", "1.18.1", "1.19.1", "1.20.0");
+ return Arrays.asList("1.17.2");
Review Comment:
Sorry, I do it for test and will modify back later.
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723162468
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java:
##########
@@ -147,17 +146,27 @@ private void applySchemaChange(
try {
metadataApplier.applySchemaChange(changeEvent);
LOG.debug("Applied schema change {} to table {}.", changeEvent, tableId);
+ schemaManager.applyEvolvedSchemaChange(changeEvent);
finishedSchemaChanges.add(changeEvent);
- } catch (SchemaEvolveException e) {
+ } catch (Exception e) {
Review Comment:
done 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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558
--
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-36094] CDC SchemaRegistryRequestHandler should throw exception which is not SchemaEvolveException [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3558:
URL: https://github.com/apache/flink-cdc/pull/3558#discussion_r1723272737
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/SchemaEvolveTest.java:
##########
@@ -1040,6 +1040,8 @@ tableId, buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
Column.physicalColumn(
"height", DOUBLE, "Height data")))));
Assertions.assertThatThrownBy(() -> processEvent(schemaOperator, addColumnEvents))
+ .cause()
+ .cause()
Review Comment:
Could we use the `rootCause()` here instead of two `cause()`?
--
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