public static Option<BaseErrorTableWriter> getErrorTableWriter(HoodieStreamer.Config cfg,
                                                                 SparkSession sparkSession,
                                                                 TypedProperties props,
                                                                 HoodieSparkEngineContext hoodieSparkContext,
                                                                 FileSystem fileSystem) {
    String errorTableWriterClass = props.getString(ERROR_TABLE_WRITE_CLASS.key());
    ValidationUtils.checkState(!StringUtils.isNullOrEmpty(errorTableWriterClass),
        "Missing error table config " + ERROR_TABLE_WRITE_CLASS);

    Class<?>[] argClassArr = new Class[] {HoodieStreamer.Config.class,
        SparkSession.class, TypedProperties.class, HoodieSparkEngineContext.class,
        FileSystem.class};
    String errMsg = "Unable to instantiate ErrorTableWriter with arguments type "
        + Arrays.toString(argClassArr);
    ValidationUtils.checkArgument(
        ReflectionUtils.hasConstructor(BaseErrorTableWriter.class.getName(), argClassArr, false),
        errMsg);

    try {
      return Option.of((BaseErrorTableWriter) ReflectionUtils.getClass(errorTableWriterClass)
          .getConstructor(argClassArr)
          .newInstance(cfg, sparkSession, props, hoodieSparkContext, fileSystem));
    } catch (NoSuchMethodException | InvocationTargetException | InstantiationException
             | IllegalAccessException e) {
      throw new HoodieException(errMsg, e);
    }
  }
@Test
  public void testGetErrorTableWriter() {
    SparkSession sparkSession = Mockito.mock(SparkSession.class);
    HoodieSparkEngineContext sparkContext = Mockito.mock(HoodieSparkEngineContext.class);
    FileSystem fileSystem = Mockito.mock(FileSystem.class);

    TypedProperties props = new TypedProperties();
    // No error table writer config
    assertThrows(IllegalArgumentException.class,
        () -> ErrorTableUtils.getErrorTableWriter(
            new HoodieStreamer.Config(), sparkSession, props, sparkContext, fileSystem));

    // Empty error table writer config
    props.put("hoodie.errortable.write.class", StringUtils.EMPTY_STRING);
    assertThrows(IllegalStateException.class,
        () -> ErrorTableUtils.getErrorTableWriter(
            new HoodieStreamer.Config(), sparkSession, props, sparkContext, fileSystem));

    // Proper error table writer config
    props.put("hoodie.errortable.write.class", TestErrorTable.class.getName());
    assertTrue(ErrorTableUtils.getErrorTableWriter(
        new HoodieStreamer.Config(), sparkSession, props, sparkContext, fileSystem).get() instanceof TestErrorTable);

    // Wrong error table writer config
    props.put("hoodie.errortable.write.class", HoodieConfig.class.getName());
    assertThrows(HoodieException.class,
        () -> ErrorTableUtils.getErrorTableWriter(
            new HoodieStreamer.Config(), sparkSession, props, sparkContext, fileSystem));
  }