在所有 Kafka 消费者结束后完成 Flink 程序

Question

我在这里设置了一个最小的示例，其中我有来自 N Kakfa 主题的 N 个流（在下面的示例中为 100 个）。

我想在看到“EndofStream”消息时完成每个流。 当所有流都完成后，我希望 Flink 程序能够优雅地完成。
当并行度设置为 1 时会出现这种情况，但通常不会发生。

从Another question看来，kafka消费者组的线程似乎不是所有的都结束了。

Others 建议抛出异常。但是，程序将在第一个异常处终止，并且不会等待所有流完成。

我还添加了一个最小的 python 程序来向 kafka 主题添加消息以实现可重复性。请在每个程序中填写<IP>:<PORT>。

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        String outputPath = "file://" + System.getProperty("user.dir") + "/out/output";

        Properties kafkaProps = null;
        kafkaProps = new Properties();
        String brokers = "<IP>:<PORT>";
        kafkaProps.setProperty("bootstrap.servers", brokers);
        kafkaProps.setProperty("auto.offset.reset", "earliest");


        ArrayList<FlinkKafkaConsumer<String>> consumersList = new ArrayList<FlinkKafkaConsumer<String>>();
        ArrayList<DataStream<String>> streamList = new ArrayList<DataStream<String>>();

        for (int i = 0; i < 100; i++) {
            consumersList.add(new FlinkKafkaConsumer<String>(Integer.toString(i),
                        new SimpleStringSchema() {
                            @Override
                            public boolean isEndOfStream(String nextElement) {
                                if (nextElement.contains("EndofStream")) {
                                    // throw new RuntimeException("End of Stream");       
                                    return true;
                                } else { 
                                    return false;
                                }
                            }
                        }
                        , kafkaProps));
            consumersList.get(i).setStartFromEarliest();
            streamList.add(env.addSource(consumersList.get(i)));
            streamList.get(i).writeAsText(outputPath + Integer.toString(i), WriteMode.OVERWRITE);
        }

        // execute program
        env.execute("Flink Streaming Java API Skeleton");

Python 3 程序

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='<IP>:<PORT>')

for i in range(100): # Channel Number
    for j in range(100): # Message Number
        message = "Message: " + str(j) + " going on channel: " + str(i)
        producer.send(str(i), str.encode(message))

    message = "EndofStream on channel: " + str(i)
    producer.send(str(i), str.encode(message))


producer.flush()

更改这一行：streamList.add(env.addSource(consumersList.get(i))); 到 streamList.add(env.addSource(consumersList.get(i)).setParallelism(1)); 也可以完成这项工作，但 Flink 会将所有消费者放置到同一台物理机器上。

我希望消费者也被分发。

flink-conf.yaml

parallelism.default: 2
cluster.evenly-spread-out-slots: true

最后的手段是将每个主题写入单独的文件并使用文件作为源而不是 kafka 消费者。
最终目标是测试 flink 需要多少时间来处理某些程序的某些工作负载。

Answer 1

使用FlinkKafkaConsumerBase 中的cancel 方法，它是FlinkKafkaConsumer 的父类。

public void cancel() 从接口复制的描述：SourceFunction 取消源。大多数来源将在内部有一个while循环 SourceFunction.run(SourceContext) 方法。实施需要 确保源将在此方法之后跳出该循环 叫做。一个典型的模式是有一个“volatile boolean” 在此方法中设置为 false 的 isRunning" 标志。该标志是 检查循环条件。

当一个源被取消时，正在执行的线程也将 中断（通过 Thread.interrupt()）。中断发生 严格在调用此方法之后，因此任何中断 handler 可以依赖这个方法已经完成的事实。它是 使通过此方法更改的任何标志“易变”的良好做法，在 为了保证这种方法的效果对任何人的可见性 中断处理程序。

指定者：接口SourceFunction中的cancel

---

你是对的。必须使用SimpleStringSchema。这是基于这个答案https://stackoverflow.com/a/44247452/2096986。看看这个例子。首先，我发送了字符串Flink code we saw also works in a cluster，Kafka 消费者消费了该消息。然后我发送SHUTDOWNDDDDDDD 也没有影响完成流。最后，我发送了SHUTDOWN，流作业完成。查看程序下方的日志。

package org.sense.flink.examples.stream.kafka;

import java.util.Properties;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class KafkaConsumerQuery {

    public KafkaConsumerQuery() throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");

        FlinkKafkaConsumer myConsumer = new FlinkKafkaConsumer(java.util.regex.Pattern.compile("test"),
                new MySimpleStringSchema(), properties);

        DataStream<String> stream = env.addSource(myConsumer);
        stream.print();

        System.out.println("Execution plan >>>\n" + env.getExecutionPlan());
        env.execute(KafkaConsumerQuery.class.getSimpleName());
    }

    private static class MySimpleStringSchema extends SimpleStringSchema {
        private static final long serialVersionUID = 1L;
        private final String SHUTDOWN = "SHUTDOWN";

        @Override
        public String deserialize(byte[] message) {

            return super.deserialize(message);
        }

        @Override
        public boolean isEndOfStream(String nextElement) {
            if (SHUTDOWN.equalsIgnoreCase(nextElement)) {
                return true;
            }
            return super.isEndOfStream(nextElement);
        }
    }

    public static void main(String[] args) throws Exception {
        new KafkaConsumerQuery();
    }
}

日志：

2020-07-02 16:39:59,025 INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-8, groupId=test] Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)
3> Flink code we saw also works in a cluster. To run this code in a cluster
3> SHUTDOWNDDDDDDD
2020-07-02 16:40:27,973 INFO  org.apache.flink.runtime.taskmanager.Task                     - Source: Custom Source -> Sink: Print to Std. Out (3/4) (5f47c2b3f55c5eb558484d49fb1fcf0e) switched from RUNNING to FINISHED.
2020-07-02 16:40:27,973 INFO  org.apache.flink.runtime.taskmanager.Task                     - Freeing task resources for Source: Custom Source -> Sink: Print to Std. Out (3/4) (5f47c2b3f55c5eb558484d49fb1fcf0e).
2020-07-02 16:40:27,974 INFO  org.apache.flink.runtime.taskmanager.Task                     - Ensuring all FileSystem streams are closed for task Source: Custom Source -> Sink: Print to Std. Out (3/4) (5f47c2b3f55c5eb558484d49fb1fcf0e) [FINISHED]
2020-07-02 16:40:27,975 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Un-registering task and sending final execution state FINISHED to JobManager for task Source: Custom Source -> Sink: Print to Std. Out (3/4) 5f47c2b3f55c5eb558484d49fb1fcf0e.
2020-07-02 16:40:27,979 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph        - Source: Custom Source -> Sink: Print to Std. Out (3/4) (5f47c2b3f55c5eb558484d49fb1fcf0e) switched from RUNNING to FINISHED.