@Override
  protected Pair<Option<Dataset<Row>>, String> fetchNextBatch(Option<String> lastCkptStr, long sourceLimit) {
    LOG.info("fetchNextBatch(): Input checkpoint: " + lastCkptStr);
    MessageBatch messageBatch;
    try {
      messageBatch = fetchFileMetadata();
    } catch (HoodieException e) {
      throw e;
    } catch (Exception e) {
      throw new HoodieReadFromSourceException("Failed to fetch file metadata from GCS events source", e);
    }

    if (messageBatch.isEmpty()) {
      LOG.info("No new data. Returning empty batch with checkpoint value: " + CHECKPOINT_VALUE_ZERO);
      return Pair.of(Option.empty(), CHECKPOINT_VALUE_ZERO);
    }

    Dataset<String> eventRecords = sparkSession.createDataset(messageBatch.getMessages(), Encoders.STRING());

    LOG.info("Returning checkpoint value: " + CHECKPOINT_VALUE_ZERO);

    StructType sourceSchema = UtilHelpers.getSourceSchema(schemaProvider);
    if (sourceSchema != null) {
      return Pair.of(Option.of(sparkSession.read().schema(sourceSchema).json(eventRecords)), CHECKPOINT_VALUE_ZERO);
    } else {
      return Pair.of(Option.of(sparkSession.read().json(eventRecords)), CHECKPOINT_VALUE_ZERO);
    }
  }
@Test
  public void shouldReturnEmptyOnNoMessages() {
    when(pubsubMessagesFetcher.fetchMessages()).thenReturn(Collections.emptyList());

    GcsEventsSource source = new GcsEventsSource(props, jsc, sparkSession, null,
            pubsubMessagesFetcher);

    Pair<Option<Dataset<Row>>, String> expected = Pair.of(Option.empty(), "0");
    Pair<Option<Dataset<Row>>, String> dataAndCheckpoint = source.fetchNextBatch(Option.of("0"), 100);

    assertEquals(expected, dataAndCheckpoint);
  }