@Override
    public void open(Map<String, Object> map, SinkContext sinkContext) throws Exception {
        try {
            val configV2 = InfluxDBSinkConfig.load(map, sinkContext);
            configV2.validate();
            sink = new InfluxDBSink();
        } catch (Exception e) {
            try {
                val configV1 = org.apache.pulsar.io.influxdb.v1.InfluxDBSinkConfig.load(map, sinkContext);
                configV1.validate();
                sink = new org.apache.pulsar.io.influxdb.v1.InfluxDBGenericRecordSink();
            } catch (Exception e1) {
                throw new Exception("For InfluxDB V2: \n" + e.toString() + "\n"
                + "For InfluxDB V1: \n" + e1.toString());
            }
        }
       sink.open(map, sinkContext);
    }
@Test(expectedExceptions = Exception.class,
            expectedExceptionsMessageRegExp = "For InfluxDB V2:.*")
    public void openInvalidInfluxConfig() throws Exception {
        InfluxDBGenericRecordSink sink = new InfluxDBGenericRecordSink();
        sink.open(new HashMap<>(), mock(SinkContext.class));
    }