ElasticsearchInputSplit[] createInputSplitsInternal(String index, String type, C client, int minNumSplits);

SearchResponse search(C client, SearchRequest searchRequest) throws IOException;

SearchResponse scroll(C client, SearchScrollRequest searchScrollRequest) throws IOException;

void close(C client) throws IOException;

//for Elasticsearch5ApiCallBridge
@Override
public SearchResponse search(TransportClient client, SearchRequest searchRequest) throws IOException {
  /how to construct SearchRequest/convert elasticsearchBuilder to SearchRequestBuiler
  return client.search(searchRequest).actionGet();
}

//for Elasticsearch6ApiCallBridge and Elasticsearch7ApiCallBridge
@Override
public SearchResponse search(RestHighLevelClient client, SearchRequest searchRequest) throws IOException {
    //convert elasticsearchBuilder to SearchRequest
  return client.search(searchRequest);
}

CREATE TABLE es_table (
   ...
  ) WITH (
   'connector' = 'elasticsearch-7',
   'hosts' = 'http://localhost:9092',
   'index' = 'MyIndex',
   'document-type' = 'MyType'
   'scan.scroll.max-size'= '1000',
   'scan.scroll.timeout'= '1000'
  );


	public static final ConfigOption<Integer> SCROLL_MAX_SIZE_OPTION =
		ConfigOptions.key("scan.scroll.max-size")
			.intType()
			.noDefaultValue()
			.withDescription("Maximum number of hits to be returned with each Elasticsearch scroll request");

	public static final ConfigOption<Duration> SCROLL_TIMEOUT_OPTION =
		ConfigOptions.key("scan.scroll.timeout")
			.durationType()
			.noDefaultValue()
			.withDescription("Amount of time Elasticsearch will keep the search context alive for scroll requests");

		Map<String, String> userConfig = new HashMap<>();
		userConfig.put("cluster.name", CLUSTER_NAME);
		DataType dataType = ROW(FIELD("data", STRING()));
		RowType schema = (RowType) dataType.getLogicalType();

		// pass on missing field
		DeserializationSchema<RowData> deserializationSchema = new JsonRowDataDeserializationSchema(
			schema, new RowDataTypeInfo(schema), false, false);

		ElasticSearchInputFormatBase inputFormat = createElasticsearchInputFormat(
			userConfig,
			(DeserializationSchema<T>) deserializationSchema,
			null,	//the fields user want
			"elasticsearch-sink-test-index",//index
			"flink-es-test-type", //type
			1000,  //scrollTimeout
			10  //scrollMaxSize
		);
		DataStream<RowTypeInfo> dataStream = env.createInput(inputFormat);
		dataStream.print();
		env.execute("Elasticsearch Source Test");

		// which is in eval method
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.fetchSource(fieldNames, null);

		for (int i = 0; i < fieldNames.length; i++) {
			looupCondition.must(new TermQueryBuilder(keyNames[i], keys[i]));
		}
		searchSourceBuilder.query(looupCondition);
		searchRequest.source(searchSourceBuilder);

 // cache settings
  final long cacheMaxSize;
  final long cacheExpireMs;

		SearchRequest searchRequest = new SearchRequest(index);
		if (type == null) {
			searchRequest.types(Strings.EMPTY_ARRAY);
		} else {
			searchRequest.types(type);
		}
		this.scroll = new Scroll(TimeValue.timeValueMinutes(scrollTimeout));
		searchRequest.scroll(scroll);
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		int size;
		if (limit > 0) {
			size = (int) Math.min(limit, scrollSize);
		} else {
			size = scrollSize;
		}
		//es scroll size default value is 10
		searchSourceBuilder.size(size);
		searchSourceBuilder.fetchSource(fieldNames, null);
		if (predicate != null) {
			searchSourceBuilder.query(predicate);
		} else {
			searchSourceBuilder.query(QueryBuilders.matchAllQuery());
		}
		searchRequest.source(searchSourceBuilder);
		searchRequest.preference("_shards:" + split.getShard());