Skip to content

[FLINK-36076] Set isSchemaChangeApplying as isSchemaChangeApplying for thread safe. #3556

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leonardBang merged 3 commits into from
Aug 19, 2024
Merged

[FLINK-36076] Set isSchemaChangeApplying as isSchemaChangeApplying for thread safe. #3556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
480a9a3
[FLINK-36076] Set isSchemaChangeApplying as isSchemaChangeApplying fo…
loserwang1024 Aug 19, 2024
0ff08b7
[FLINK-36076] Set isSchemaChangeApplying as isSchemaChangeApplying fo…
loserwang1024 Aug 19, 2024
178d9fe
[FLINK-36076] Set isSchemaChangeApplying as volatile for thread safe.
loserwang1024 Aug 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,8 @@ public PipelineExecution compose(PipelineDef pipelineDef) {
// Build Source Operator
DataSourceTranslator sourceTranslator = new DataSourceTranslator();
DataStream<Event> stream =
sourceTranslator.translate(pipelineDef.getSource(), env, pipelineDef.getConfig());
sourceTranslator.translate(
pipelineDef.getSource(), env, pipelineDef.getConfig(), parallelism);

// Build PreTransformOperator for processing Schema Event
TransformTranslator transformTranslator = new TransformTranslator();
Expand Down
7 changes: 4 additions & 3 deletions ...er/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.factories.DataSourceFactory;
import org.apache.flink.cdc.common.factories.FactoryHelper;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.source.DataSource;
import org.apache.flink.cdc.common.source.EventSourceProvider;
import org.apache.flink.cdc.common.source.FlinkSourceFunctionProvider;
Expand All @@ -41,12 +40,14 @@
public class DataSourceTranslator {

public DataStreamSource<Event> translate(
SourceDef sourceDef, StreamExecutionEnvironment env, Configuration pipelineConfig) {
SourceDef sourceDef,
StreamExecutionEnvironment env,
Configuration pipelineConfig,
int sourceParallelism) {
// Create data source
DataSource dataSource = createDataSource(sourceDef, env, pipelineConfig);

// Get source provider
final int sourceParallelism = pipelineConfig.get(PipelineOptions.PIPELINE_PARALLELISM);
EventSourceProvider eventSourceProvider = dataSource.getEventSourceProvider();
if (eventSourceProvider instanceof FlinkSourceProvider) {
// Source
Expand Down
2 changes: 1 addition & 1 deletion ...g/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ public class SchemaRegistryRequestHandler implements Closeable {
private final Set<Integer> flushedSinkWriters;

/** Status of the execution of current schema change request. */
private boolean isSchemaChangeApplying;
private volatile boolean isSchemaChangeApplying;
/** Executor service to execute schema change. */
private final ExecutorService schemaChangeThreadPool;

Expand Down