@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
    public void openInfluxV2() throws Exception {
        Map<String, Object> map = new HashMap();
        map.put("influxdbUrl", "http://localhost:9999");
        map.put("token", "xxxx");
        map.put("organization", "example-org");
        map.put("bucket", "example-bucket");

        InfluxDBGenericRecordSink sink = new InfluxDBGenericRecordSink();
        try {
            sink.open(map, mock(SinkContext.class));
        } catch (InfluxDBIOException e) {
            // Do nothing
        }
        assertTrue(sink.sink instanceof InfluxDBSink);
    }