线束还包含一个 .withFlinkSourceFunction() 允许使用任何flink的方法 SourceFunction 作为入口。
您可以创建自己的源函数来生成有限的元素集合，例如：

class FiniteSource<T extends Serializable> implements SourceFunction<T> {
    private final List<T> items;

    FiniteSource(List<T> items) {
      this.items = items;
    }

    @Override
    public void run(SourceContext<T> sourceContext) {
      for (T item : items) {
        sourceContext.collect(item);
      }
    }

    @Override
    public void cancel() {}
  }

然后，可以按以下方式修改线束示例：

FiniteSource<MyInputMessage> finiteSource = new FiniteSource<>(
            Arrays.asList(
                    new MyInputMessage("user-1", "hello"),
                    new MyInputMessage("user-2", "world")));

    Harness harness =
        new Harness()
            .withKryoMessageSerializer()
            .withFlinkSourceFunction(MyConstants.REQUEST_INGRESS,finiteSource)
            .withPrintingEgress(MyConstants.RESULT_EGRESS);

    harness.start();

在一个入口中产生两条输入消息后，这应该终止。如果您认为这是一个常见的要求，那么我鼓励您在flink邮件列表中提出这一点，我相信那里的友好社区会很乐意接受您的反馈意见，更愿意做出贡献；）
有没有人找到了解决上述问题的方法，或者发现了一种完全不同的方法，使用的工具不是线束？
对于ci/cd管道，我建议检查基于测试容器的e2e测试(例如这个）