ElasticSourceBuilder { @Nonnull public BatchSource<T> build() { requireNonNull(clientFn, "clientFn must be set"); requireNonNull(searchRequestFn, "searchRequestFn must be set"); requireNonNull(mapToItemFn, "mapToItemFn must be set"); ElasticSourceConfiguration<T> configuration = new ElasticSourceConfiguration<>( restHighLevelClientFn(clientFn), searchRequestFn, optionsFn, mapToItemFn, slicing, coLocatedReading, scrollKeepAlive, retries ); ElasticSourcePMetaSupplier<T> metaSupplier = new ElasticSourcePMetaSupplier<>(configuration); return Sources.batchFromProcessor(DEFAULT_NAME, metaSupplier); }  @Nonnull BatchSource<T> build(); @Nonnull ElasticSourceBuilder<T> clientFn(@Nonnull SupplierEx<RestClientBuilder> clientFn); @Nonnull ElasticSourceBuilder<T> searchRequestFn(@Nonnull SupplierEx<SearchRequest> searchRequestFn); @Nonnull @SuppressWarnings("unchecked") ElasticSourceBuilder<T_NEW> mapToItemFn(@Nonnull FunctionEx<? super SearchHit, T_NEW> mapToItemFn); @Nonnull ElasticSourceBuilder<T> optionsFn(@Nonnull FunctionEx<? super ActionRequest, RequestOptions> optionsFn); @Nonnull ElasticSourceBuilder<T> enableSlicing(); @Nonnull ElasticSourceBuilder<T> enableCoLocatedReading(); @Nonnull ElasticSourceBuilder<T> scrollKeepAlive(@Nonnull String scrollKeepAlive); @Nonnull ElasticSourceBuilder<T> retries(int retries);  }
@Test public void when_createElasticSourceUsingBuilder_then_sourceHasCorrectName() { BatchSource<Object> source = builderWithRequiredParams() .build(); assertThat(source.name()).isEqualTo("elasticSource"); }