@Override
    public void open(Map<String, Object> config, SinkContext ctx) throws Exception {
        kafkaSinkConfig = PulsarKafkaConnectSinkConfig.load(config);
        Objects.requireNonNull(kafkaSinkConfig.getTopic(), "Kafka topic is not set");
        Preconditions.checkArgument(ctx.getSubscriptionType() == SubscriptionType.Failover
                || ctx.getSubscriptionType() == SubscriptionType.Exclusive,
                "Source must run with Exclusive or Failover subscription type");
        topicName = kafkaSinkConfig.getTopic();
        unwrapKeyValueIfAvailable = kafkaSinkConfig.isUnwrapKeyValueIfAvailable();
        sanitizeTopicName = kafkaSinkConfig.isSanitizeTopicName();
        collapsePartitionedTopics = kafkaSinkConfig.isCollapsePartitionedTopics();
        useOptionalPrimitives = kafkaSinkConfig.isUseOptionalPrimitives();

        useIndexAsOffset = kafkaSinkConfig.isUseIndexAsOffset();
        maxBatchBitsForOffset = kafkaSinkConfig.getMaxBatchBitsForOffset();
        Preconditions.checkArgument(maxBatchBitsForOffset <= 20,
                "Cannot use more than 20 bits for maxBatchBitsForOffset");

        String kafkaConnectorFQClassName = kafkaSinkConfig.getKafkaConnectorSinkClass();
        kafkaSinkConfig.getKafkaConnectorConfigProperties().forEach(props::put);

        Class<?> clazz = Class.forName(kafkaConnectorFQClassName);
        connector = (SinkConnector) clazz.getConstructor().newInstance();

        Class<? extends Task> taskClass = connector.taskClass();
        sinkContext = new PulsarKafkaSinkContext();
        connector.initialize(sinkContext);
        connector.start(Maps.fromProperties(props));

        List<Map<String, String>> configs = connector.taskConfigs(1);
        Preconditions.checkNotNull(configs);
        Preconditions.checkArgument(configs.size() == 1);

        // configs may contain immutable/unmodifiable maps
        configs = configs.stream()
                .map(HashMap::new)
                .collect(Collectors.toList());

        configs.forEach(x -> {
            x.put(PulsarKafkaWorkerConfig.OFFSET_STORAGE_TOPIC_CONFIG, kafkaSinkConfig.getOffsetStorageTopic());
        });
        task = (SinkTask) taskClass.getConstructor().newInstance();
        taskContext =
                new PulsarKafkaSinkTaskContext(configs.get(0), ctx, task::open, kafkaName -> {
                    if (sanitizeTopicName) {
                        String pulsarTopicName = desanitizedTopicCache.getIfPresent(kafkaName);
                        if (log.isDebugEnabled()) {
                            log.debug("desanitizedTopicCache got: kafkaName: {}, pulsarTopicName: {}",
                                    kafkaName, pulsarTopicName);
                        }
                        return pulsarTopicName != null ? pulsarTopicName : kafkaName;
                    } else {
                        return kafkaName;
                    }
                });
        task.initialize(taskContext);
        task.start(configs.get(0));

        maxBatchSize = kafkaSinkConfig.getBatchSize();
        lingerMs = kafkaSinkConfig.getLingerTimeMs();

        isRunning = true;
        scheduledExecutor.scheduleWithFixedDelay(() ->
                this.flushIfNeeded(true), lingerMs, lingerMs, TimeUnit.MILLISECONDS);

        log.info("Kafka sink started : {}.", props);
    }
@Test
    public void subscriptionTypeTest() throws Exception {
        try (KafkaConnectSink sink = new KafkaConnectSink()) {
            log.info("Failover is allowed");
            sink.open(props, context);
        }

        when(context.getSubscriptionType()).thenReturn(SubscriptionType.Exclusive);
        try (KafkaConnectSink sink = new KafkaConnectSink()) {
            log.info("Exclusive is allowed");
            sink.open(props, context);
        }

        when(context.getSubscriptionType()).thenReturn(SubscriptionType.Key_Shared);
        try (KafkaConnectSink sink = new KafkaConnectSink()) {
            log.info("Key_Shared is not allowed");
            sink.open(props, context);
            fail("expected exception");
        } catch (IllegalArgumentException iae) {
            // pass
        }

        when(context.getSubscriptionType()).thenReturn(SubscriptionType.Shared);
        try (KafkaConnectSink sink = new KafkaConnectSink()) {
            log.info("Shared is not allowed");
            sink.open(props, context);
            fail("expected exception");
        } catch (IllegalArgumentException iae) {
            // pass
        }

        when(context.getSubscriptionType()).thenReturn(null);
        try (KafkaConnectSink sink = new KafkaConnectSink()) {
            log.info("Type is required");
            sink.open(props, context);
            fail("expected exception");
        } catch (IllegalArgumentException iae) {
            // pass
        }

    }