Flink流未完成

Question

我正在使用kafka和elasticsearch设置flink流处理器。我想重播我的数据，但当我将并行度设置为1以上时，它不会完成程序，我认为这是因为Kafka流只看到一条消息，将其标识为流的结尾。

    public CustomSchema(Date _endTime) {
        endTime = _endTime;
    }

@Override
    public boolean isEndOfStream(CustomTopicWrapper nextElement) {
        if (this.endTime != null && nextElement.messageTime.getTime() >= this.endTime.getTime()) {
            return true;
        }
        return false;
    }

有没有办法告诉flink消费群中的所有线程在一个线程完成后立即结束？

Answer 1

如果实现了自己的SourceFunction，请使用Flink SourceFunction中的示例所示的取消方法。FlinkKafkaConsumerBase类还具有cancel方法。