Akka流停止处理数据

Question

当我运行下面的流时，一旦流运行，就不会收到任何后续数据。

    final long HOUR = 3600000;
    final long PAST_HOUR = System.currentTimeMillis()-HOUR;

private final static ActorSystem actorSystem = ActorSystem.create(Behaviors.empty(), "as");

protected static ElasticsearchParams constructElasticsearchParams(
        String indexName, String typeName, ApiVersion apiVersion) {
    if (apiVersion == ApiVersion.V5) {
        return ElasticsearchParams.V5(indexName, typeName);
    } else if (apiVersion == ApiVersion.V7) {
        return ElasticsearchParams.V7(indexName);
    }
    else {
        throw new IllegalArgumentException("API version " + apiVersion + " is not supported");
    }
}

    String queryStr = "{ \"bool\": {  \"must\" : [{\"range\" : {"+
            "\"timestamp\" : { "+
            "\"gte\" : "+PAST_HOUR
            +" }} }]}} ";

    ElasticsearchConnectionSettings connectionSettings =
            ElasticsearchConnectionSettings.create("****")
                    .withCredentials("****", "****");

    ElasticsearchSourceSettings sourceSettings =
            ElasticsearchSourceSettings.create(connectionSettings)
                    .withApiVersion(ApiVersion.V7);

    Source<ReadResult<Stats>, NotUsed> dataSource =
            ElasticsearchSource.typed(
                    constructElasticsearchParams("data", "_doc", ApiVersion.V7),
                    queryStr,
                    sourceSettings,
                    Stats.class);

    dataSource.buffer(10000, OverflowStrategy.backpressure());
    dataSource.backpressureTimeout(Duration.ofSeconds(1));

    dataSource
            .log("error")
            .runWith(Sink.foreach(a -> System.out.println(a)), actorSystem);

产生输出：

ReadResult(id=1656107389556,source=Stats(size=0.09471),version=)

数据不断地写入索引数据，但是一旦开始，该流就不会对其进行处理。该流不应该从上游来源处理数据吗？在这种情况下，上游源是一个名为数据的弹性索引。

我尝试修改查询以匹配所有文档：

String queryStr =  "{\"match_all\": {}}";

但是结果相同。

Answer 1

Elasticsearch源不会连续运行。它启动搜索，管理分页（使用批量API）并流式播放；当Elasticsearch报告不再完成结果时，它将完成。

您可以做类似的事情

Source.repeat(Done).flatMapConcat(done -> ElasticsearchSource.typed(...))

，这些事情将在上一个结束后立即进行新的搜索。请注意，下游过滤重复物将是责任。