public static Optional<MaximumLagFilter> create(
      final Optional<LagReportingAgent> lagReportingAgent,
      final RoutingOptions routingOptions,
      final List<KsqlHostInfo> hosts,
      final String queryId,
      final String storeName,
      final int partition
  ) {
    if (!lagReportingAgent.isPresent()) {
      return Optional.empty();
    }
    final QueryStateStoreId queryStateStoreId = QueryStateStoreId.of(queryId, storeName);
    final ImmutableMap<KsqlHostInfo, Optional<LagInfoEntity>> lagByHost = hosts.stream()
        .collect(ImmutableMap.toImmutableMap(
            Function.identity(),
            host -> lagReportingAgent.get().getLagInfoForHost(
                host,
                queryStateStoreId,
                partition)));
    final OptionalLong maxEndOffset = lagByHost.values().stream()
        .filter(Optional::isPresent)
        .map(Optional::get)
        .mapToLong(LagInfoEntity::getEndOffsetPosition)
        .max();
    return Optional.of(new MaximumLagFilter(routingOptions, lagByHost, maxEndOffset));
  }
@Test
  public void filter_lagReportingDisabled() {
    // When:
    Optional<MaximumLagFilter> filterOptional = MaximumLagFilter.create(
        Optional.empty(), routingOptions, HOSTS, APPLICATION_ID, STATE_STORE, PARTITION);

    // Then:
    assertFalse(filterOptional.isPresent());
  }