@Override
    public CloseableIterator<ColumnarBatch> readJsonFiles(
            CloseableIterator<FileStatus> scanFileIter,
            StructType physicalSchema,
            Optional<Predicate> predicate) {

        return new CloseableIterator<>() {
            private String currentFile;
            // index of the current line being read from the current read json list, -1 means no line is read yet
            private int currentReadLine = -1;
            private List<JsonNode> currentReadJsonList = Lists.newArrayList();

            @Override
            public void close() {
                Utils.closeCloseables(scanFileIter);
                currentReadLine = -1;
                currentReadJsonList = null;
            }

            @Override
            public boolean hasNext() {
                if (hasNextToConsume()) {
                    return true; // we have un-consumed last read line
                }

                // There is no file in reading or the current file being read has no more data
                // initialize the next file reader or return false if there are no more files to
                // read.
                try {
                    tryGetNextFileJson();
                } catch (Exception ex) {
                    throw new KernelEngineException(
                            format("Error reading JSON file: %s", currentFile), ex);
                }

                return hasNextToConsume();
            }

            private boolean hasNextToConsume() {
                return currentReadLine != -1 && !currentReadJsonList.isEmpty() && currentReadLine < currentReadJsonList.size();
            }

            @Override
            public ColumnarBatch next() {
                try (Timer ignored = Tracers.watchScope(Tracers.get(), EXTERNAL, "DeltaLakeJsonHandler.JsonToColumnarBatch")) {
                    if (!hasNextToConsume()) {
                        throw new NoSuchElementException();
                    }

                    List<Row> rows = new ArrayList<>();
                    int currentBatchSize = 0;
                    do {
                        // hasNext already reads the next file and keeps it in member variable `cachedJsonList`
                        JsonNode jsonNode = currentReadJsonList.get(currentReadLine);
                        Row row = new io.delta.kernel.defaults.internal.data.DefaultJsonRow(
                                (ObjectNode) jsonNode, physicalSchema);
                        rows.add(row);
                        currentBatchSize++;
                        currentReadLine++;
                    } while (currentBatchSize < maxBatchSize && hasNext());

                    return new io.delta.kernel.defaults.internal.data.DefaultRowBasedColumnarBatch(
                            physicalSchema, rows);
                }
            }

            private void tryGetNextFileJson() throws ExecutionException, IOException {
                if (scanFileIter.hasNext()) {
                    currentFile = scanFileIter.next().getPath();
                    Path filePath = new Path(currentFile);
                    if (filePath.getName().equals(LAST_CHECKPOINT_FILE_NAME)) {
                        // can not read last_checkpoint file from cache
                        currentReadJsonList = readJsonFile(currentFile, hadoopConf);
                    } else {
                        currentReadJsonList = jsonCache.get(currentFile);
                    }
                    currentReadLine = 0;
                }
            }
        };
    }
@Test
    public void testReadLastCheckPoint() {
        String path = deltaLakePath + "/_last_checkpoint";
        DeltaLakeJsonHandler deltaLakeJsonHandler = new DeltaLakeJsonHandler(hdfsConfiguration, jsonCache);

        StructType readSchema = CheckpointMetaData.READ_SCHEMA;
        FileStatus fileStatus = FileStatus.of(path, 0, 0);
        try (CloseableIterator<ColumnarBatch> jsonIter = deltaLakeJsonHandler.readJsonFiles(
                Utils.singletonCloseableIterator(fileStatus), readSchema, Optional.empty())) {
            Optional<Row> checkpointRow = InternalUtils.getSingularRow(jsonIter);
            Assert.assertTrue(checkpointRow.isPresent());
            Row row = checkpointRow.get();
            Assert.assertEquals(row.getLong(0), 30);
        } catch (IOException e) {
            Assert.fail();
        }

        Assert.assertTrue(jsonCache.asMap().isEmpty());
    }