Shawn-Hx commented on code in PR #3801: URL: https://github.com/apache/flink-cdc/pull/3801#discussion_r1891283182
########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.partitioning; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.function.HashFunction; +import org.apache.flink.cdc.common.function.HashFunctionProvider; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.utils.SchemaUtils; +import org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator; +import org.apache.flink.cdc.runtime.serializer.event.EventSerializer; +import org.apache.flink.runtime.state.StateSnapshotContext; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.ChainingStrategy; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; + +/** + * Operator for processing events from {@link SchemaOperator} before {@link EventPartitioner} with Review Comment: Comments should be updated. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java: ########## @@ -0,0 +1,399 @@ +/* + * 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.common; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.annotation.VisibleForTesting; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior; +import org.apache.flink.cdc.common.route.RouteRule; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.sink.MetadataApplier; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.FlushSuccessEvent; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.GetEvolvedSchemaRequest; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.GetEvolvedSchemaResponse; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.GetOriginalSchemaRequest; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.GetOriginalSchemaResponse; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.SinkWriterRegisterEvent; +import org.apache.flink.cdc.runtime.operators.sink.SchemaEvolutionClient; +import org.apache.flink.runtime.operators.coordination.CoordinationRequest; +import org.apache.flink.runtime.operators.coordination.CoordinationRequestHandler; +import org.apache.flink.runtime.operators.coordination.CoordinationResponse; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.function.ThrowingRunnable; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeoutException; +import java.util.function.BooleanSupplier; + +import static org.apache.flink.cdc.runtime.operators.schema.common.event.common.CoordinationResponseUtils.wrap; + +/** + * An abstract centralized schema registry that accepts requests from {@link SchemaEvolutionClient}. + * A legit schema registry should be able to: + * + * <ul> + * <li>Handle schema retrieval requests from {@link SchemaEvolutionClient}s + * <li>Accept and trace sink writers' registering events + * <li>Snapshot & Restore its state during checkpoints + * </ul> + * + * <br> + * These abilities are done by overriding abstract methods of {@link SchemaRegistry}. All methods + * will run in given {@link ExecutorService} asynchronously except {@code SchemaRegistry#restore}. + */ +@Internal +public abstract class SchemaRegistry implements OperatorCoordinator, CoordinationRequestHandler { + + private static final Logger LOG = LoggerFactory.getLogger(SchemaRegistry.class); + + // ------------------------- + // Static fields that got bind after construction + // ------------------------- + protected final OperatorCoordinator.Context context; + protected final String operatorName; + protected final ExecutorService coordinatorExecutor; + protected final MetadataApplier metadataApplier; + protected final Duration rpcTimeout; + protected final List<RouteRule> routingRules; + protected final SchemaChangeBehavior behavior; + + // ------------------------- + // Dynamically initialized transient fields (after coordinator starts) + // ------------------------- + protected transient int currentParallelism; + protected transient Set<Integer> activeSinkWriters; + protected transient Map<Integer, SubtaskGateway> subtaskGatewayMap; Review Comment: `subtaskGatewayMap` is not used now. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java: ########## @@ -0,0 +1,240 @@ +/* + * 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.distributed; + +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.FlushEvent; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.exceptions.SchemaEvolveException; +import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior; +import org.apache.flink.cdc.common.route.RouteRule; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.utils.SchemaUtils; +import org.apache.flink.cdc.runtime.operators.schema.common.SchemaDerivator; +import org.apache.flink.cdc.runtime.operators.schema.common.TableIdRouter; +import org.apache.flink.cdc.runtime.operators.schema.common.event.common.CoordinationResponseUtils; +import org.apache.flink.cdc.runtime.operators.schema.common.event.distributed.SchemaChangeRequest; +import org.apache.flink.cdc.runtime.operators.schema.common.event.distributed.SchemaChangeResponse; +import org.apache.flink.cdc.runtime.operators.schema.common.metrics.SchemaOperatorMetrics; +import org.apache.flink.cdc.runtime.partitioning.PartitioningEvent; +import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway; +import org.apache.flink.runtime.operators.coordination.CoordinationRequest; +import org.apache.flink.runtime.operators.coordination.CoordinationResponse; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.runtime.operators.coordination.OperatorEventHandler; +import org.apache.flink.streaming.api.graph.StreamConfig; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.operators.Output; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.tasks.StreamTask; +import org.apache.flink.util.SerializedValue; + +import org.apache.flink.shaded.guava31.com.google.common.collect.HashBasedTable; +import org.apache.flink.shaded.guava31.com.google.common.collect.Table; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +/** This operator merges upstream inferred schema into a centralized Schema Registry. */ +public class SchemaOperator extends AbstractStreamOperator<Event> + implements OneInputStreamOperator<PartitioningEvent, Event>, + OperatorEventHandler, Review Comment: Can `OperatorEventHandler` be removed ? ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/event/common/CoordinationResponseUtils.java: ########## @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.flink.cdc.runtime.operators.schema.event; +package org.apache.flink.cdc.runtime.operators.schema.common.event.common; Review Comment: It's a little weird that there are two `common` in package name. It could be more clear by moving classes in `org.apache.flink.cdc.runtime.operators.schema.common.event.distributed` to `org.apache.flink.cdc.runtime.operators.schema.distributed.event`. Also for classes in `org.apache.flink.cdc.runtime.operators.schema.common.event.regular`. 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
