westphal-jan commented on a change in pull request #15140: URL: https://github.com/apache/flink/pull/15140#discussion_r594330875
########## File path: flink-connectors/flink-connector-rabbitmq2/src/main/java/org/apache/flink/connector/rabbitmq2/source/enumerator/RabbitMQSourceEnumerator.java ########## @@ -0,0 +1,120 @@ +/* + * 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.connector.rabbitmq2.source.enumerator; + +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.connector.source.SplitsAssignment; +import org.apache.flink.connector.rabbitmq2.ConsistencyMode; +import org.apache.flink.connector.rabbitmq2.RabbitMQConnectionConfig; +import org.apache.flink.connector.rabbitmq2.source.split.RabbitMQSourceSplit; +import org.apache.flink.util.FlinkRuntimeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.List; + +/** + * The source enumerator provides the source readers with the split. All source readers receive the + * same split as it only contains information about the connection and in case of exactly-once, the + * seen correlation ids. But in this case, the enumerator makes sure that at maximum one source + * reader receives the split. During exactly-once if multiple reader should be assigned a split a + * {@link FlinkRuntimeException} is thrown. + */ +public class RabbitMQSourceEnumerator + implements SplitEnumerator<RabbitMQSourceSplit, RabbitMQSourceEnumState> { + private final SplitEnumeratorContext<RabbitMQSourceSplit> context; + private final ConsistencyMode consistencyMode; + private static final Logger LOG = LoggerFactory.getLogger(RabbitMQSourceEnumerator.class); + private RabbitMQSourceSplit split; + + public RabbitMQSourceEnumerator( + SplitEnumeratorContext<RabbitMQSourceSplit> context, + ConsistencyMode consistencyMode, + RabbitMQConnectionConfig connectionConfig, + String rmqQueueName, + RabbitMQSourceEnumState enumState) { + // The enumState is not used since the enumerator has no state in this architecture. + this(context, consistencyMode, connectionConfig, rmqQueueName); + } + + public RabbitMQSourceEnumerator( + SplitEnumeratorContext<RabbitMQSourceSplit> context, + ConsistencyMode consistencyMode, + RabbitMQConnectionConfig connectionConfig, + String rmqQueueName) { + this.context = context; + this.consistencyMode = consistencyMode; + this.split = new RabbitMQSourceSplit(connectionConfig, rmqQueueName); + } + + @Override + public void start() { + LOG.info("Start rabbitmq source enumerator"); + } + + @Override + public void handleSplitRequest(int i, @Nullable String s) { + LOG.info("Split request from reader " + i); + assignSplitToReader(i, split); + } + + @Override + public void addSplitsBack(List<RabbitMQSourceSplit> list, int i) { + LOG.info("Splits returned from reader " + i); + if (list.size() == 0) { + return; + } + // Every Source Reader will only receive one splits, thus we will never get back more. + assert list.size() == 1; + split = list.get(0); Review comment: In case of exactly-once the single split gets updated with the correlation ids and therefore needs to be updated in the enumerator. For the others, you are right, it doesn't matter. We kept the exactly-once case to simplify the code. Do you have a suggestion to make it more clear? ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
