### Question:
ElasticsearchRecordHandler extends RecordHandler { @VisibleForTesting protected int getQueryBatchSize() { return QUERY_BATCH_SIZE; } ElasticsearchRecordHandler(); @VisibleForTesting protected  ElasticsearchRecordHandler(AmazonS3 amazonS3, AWSSecretsManager secretsManager, AmazonAthena amazonAthena,
                                         AwsRestHighLevelClientFactory clientFactory, long queryTimeout);   }### Answer:
@Test public void doReadRecordsSpill() throws Exception { logger.info("doReadRecordsSpill: enter"); int batchSize = handler.getQueryBatchSize(); SearchHit searchHit1[] = new SearchHit[batchSize]; for (int i = 0; i < batchSize; ++i) { searchHit1[i] = new SearchHit(i + 1); } SearchHit searchHit2[] = new SearchHit[2]; searchHit2[0] = new SearchHit(batchSize + 1); searchHit2[1] = new SearchHit(batchSize + 2); SearchHits searchHits1 = new SearchHits(searchHit1, new TotalHits(batchSize, TotalHits.Relation.EQUAL_TO), 4); SearchHits searchHits2 = new SearchHits(searchHit2, new TotalHits(2, TotalHits.Relation.EQUAL_TO), 4); when(mockResponse.getHits()) .thenReturn(searchHits1, searchHits1, searchHits2, searchHits2); Map<String, ValueSet> constraintsMap = new HashMap<>(); constraintsMap.put("myshort", SortedRangeSet.copyOf(Types.MinorType.SMALLINT.getType(), ImmutableList.of(Range.range(allocator, Types.MinorType.SMALLINT.getType(), (short) 1955, false, (short) 1972, true)), false)); ReadRecordsRequest request = new ReadRecordsRequest(fakeIdentity(), "elasticsearch", "queryId-" + System.currentTimeMillis(), new TableName("movies", "mishmash"), mapping, split, new Constraints(constraintsMap), 10_000L, 0L ); RecordResponse rawResponse = handler.doReadRecords(allocator, request); assertTrue(rawResponse instanceof RemoteReadRecordsResponse); try (RemoteReadRecordsResponse response = (RemoteReadRecordsResponse) rawResponse) { logger.info("doReadRecordsSpill: remoteBlocks[{}]", response.getRemoteBlocks().size()); assertEquals(3, response.getNumberBlocks()); int blockNum = 0; for (SpillLocation next : response.getRemoteBlocks()) { S3SpillLocation spillLocation = (S3SpillLocation) next; try (Block block = spillReader.read(spillLocation, response.getEncryptionKey(), response.getSchema())) { logger.info("doReadRecordsSpill: blockNum[{}] and recordCount[{}]", blockNum++, block.getRowCount()); logger.info("doReadRecordsSpill: {}", BlockUtils.rowToString(block, 0)); assertNotNull(BlockUtils.rowToString(block, 0)); } } } logger.info("doReadRecordsSpill: exit"); }