@Override
  public void init(Configuration config) throws HoodieException {
    try {
      this.fs = FileSystem.get(config);
    } catch (IOException e) {
      throw new HoodieException("KafkaConnectHdfsProvider initialization failed");
    }
  }
@Test
  public void testMissingPartition() throws Exception {
    Path topicPath = tempDir.resolve("topic2");
    Files.createDirectories(topicPath);
    // create regular kafka connect hdfs dirs
    new File(topicPath + "/year=2016/month=05/day=01/").mkdirs();
    new File(topicPath + "/year=2016/month=05/day=02/").mkdirs();
    // base files with missing partition
    new File(topicPath + "/year=2016/month=05/day=01/"
        + "topic1+0+100+200" + BASE_FILE_EXTENSION).createNewFile();
    new File(topicPath + "/year=2016/month=05/day=01/"
        + "topic1+2+100+200" + BASE_FILE_EXTENSION).createNewFile();
    new File(topicPath + "/year=2016/month=05/day=02/"
        + "topic1+0+201+300" + BASE_FILE_EXTENSION).createNewFile();
    final TypedProperties props = new TypedProperties();
    props.put("hoodie.streamer.checkpoint.provider.path", topicPath.toString());
    final InitialCheckPointProvider provider = new KafkaConnectHdfsProvider(props);
    provider.init(HoodieTestUtils.getDefaultStorageConf().unwrap());
    assertThrows(HoodieException.class, provider::getCheckpoint);
  }