@Override
    public ConsumerBuilder<T> loadConf(Map<String, Object> config) {
        this.conf = ConfigurationDataUtils.loadData(config, conf, ConsumerConfigurationData.class);
        return this;
    }
@Test
    public void testLoadConf() throws Exception {
        ConsumerBuilderImpl<byte[]> consumerBuilder = createConsumerBuilder();

        String jsonConf = ("{\n"
            + "    'topicNames' : [ 'new-topic' ],\n"
            + "    'topicsPattern' : 'new-topics-pattern',\n"
            + "    'subscriptionName' : 'new-subscription',\n"
            + "    'subscriptionType' : 'Key_Shared',\n"
            + "    'subscriptionProperties' : {\n"
            + "      'new-sub-prop' : 'new-sub-prop-value'\n"
            + "    },\n"
            + "    'subscriptionMode' : 'NonDurable',\n"
            + "    'receiverQueueSize' : 2,\n"
            + "    'acknowledgementsGroupTimeMicros' : 2,\n"
            + "    'maxAcknowledgmentGroupSize' : 2,\n"
            + "    'negativeAckRedeliveryDelayMicros' : 2,\n"
            + "    'maxTotalReceiverQueueSizeAcrossPartitions' : 2,\n"
            + "    'consumerName' : 'new-consumer',\n"
            + "    'ackTimeoutMillis' : 2,\n"
            + "    'tickDurationMillis' : 2,\n"
            + "    'priorityLevel' : 2,\n"
            + "    'maxPendingChunkedMessage' : 2,\n"
            + "    'autoAckOldestChunkedMessageOnQueueFull' : true,\n"
            + "    'expireTimeOfIncompleteChunkedMessageMillis' : 2,\n"
            + "    'cryptoFailureAction' : 'DISCARD',\n"
            + "    'properties' : {\n"
            + "      'new-prop' : 'new-prop-value'\n"
            + "    },\n"
            + "    'readCompacted' : true,\n"
            + "    'subscriptionInitialPosition' : 'Earliest',\n"
            + "    'patternAutoDiscoveryPeriod' : 2,\n"
            + "    'regexSubscriptionMode' : 'AllTopics',\n"
            + "    'deadLetterPolicy' : {\n"
            + "      'retryLetterTopic' : 'new-retry',\n"
            + "      'initialSubscriptionName' : 'new-dlq-sub',\n"
            + "      'deadLetterTopic' : 'new-dlq',\n"
            + "      'maxRedeliverCount' : 2\n"
            + "    },\n"
            + "    'retryEnable' : true,\n"
            + "    'autoUpdatePartitions' : false,\n"
            + "    'autoUpdatePartitionsIntervalSeconds' : 2,\n"
            + "    'replicateSubscriptionState' : true,\n"
            + "    'resetIncludeHead' : true,\n"
            + "    'batchIndexAckEnabled' : true,\n"
            + "    'ackReceiptEnabled' : true,\n"
            + "    'poolMessages' : true,\n"
            + "    'startPaused' : true,\n"
            + "    'autoScaledReceiverQueueSizeEnabled' : true\n"
            + "  }").replace("'", "\"");

        Map<String, Object> conf = new ObjectMapper().readValue(jsonConf, new TypeReference<HashMap<String,Object>>() {});

        MessageListener<byte[]> messageListener = (consumer, message) -> {};
        conf.put("messageListener", messageListener);
        ConsumerEventListener consumerEventListener = createMockConsumerEventListener();
        conf.put("consumerEventListener", consumerEventListener);
        RedeliveryBackoff negativeAckRedeliveryBackoff = MultiplierRedeliveryBackoff.builder().build();
        conf.put("negativeAckRedeliveryBackoff", negativeAckRedeliveryBackoff);
        RedeliveryBackoff ackTimeoutRedeliveryBackoff = MultiplierRedeliveryBackoff.builder().build();;
        conf.put("ackTimeoutRedeliveryBackoff", ackTimeoutRedeliveryBackoff);
        CryptoKeyReader cryptoKeyReader = DefaultCryptoKeyReader.builder().build();
        conf.put("cryptoKeyReader", cryptoKeyReader);
        MessageCrypto messageCrypto = new MessageCryptoBc("ctx2", true);
        conf.put("messageCrypto", messageCrypto);
        BatchReceivePolicy batchReceivePolicy = BatchReceivePolicy.builder().maxNumBytes(2).build();
        conf.put("batchReceivePolicy", batchReceivePolicy);
        KeySharedPolicy keySharedPolicy = KeySharedPolicy.stickyHashRange();
        conf.put("keySharedPolicy", keySharedPolicy);
        MessagePayloadProcessor payloadProcessor = createMockMessagePayloadProcessor();
        conf.put("payloadProcessor", payloadProcessor);

        consumerBuilder.loadConf(conf);

        ConsumerConfigurationData<byte[]> configurationData = consumerBuilder.getConf();
        assertEquals(configurationData.getTopicNames(), new HashSet<>(Collections.singletonList("new-topic")));
        assertEquals(configurationData.getTopicsPattern().pattern(), "new-topics-pattern");
        assertEquals(configurationData.getSubscriptionName(), "new-subscription");
        assertEquals(configurationData.getSubscriptionType(), SubscriptionType.Key_Shared);
        assertThat(configurationData.getSubscriptionProperties()).hasSize(1)
            .hasFieldOrPropertyWithValue("new-sub-prop", "new-sub-prop-value");
        assertEquals(configurationData.getSubscriptionMode(), SubscriptionMode.NonDurable);
        assertEquals(configurationData.getReceiverQueueSize(), 2);
        assertEquals(configurationData.getAcknowledgementsGroupTimeMicros(), 2);
        assertEquals(configurationData.getMaxAcknowledgmentGroupSize(), 2);
        assertEquals(configurationData.getNegativeAckRedeliveryDelayMicros(), 2);
        assertEquals(configurationData.getMaxTotalReceiverQueueSizeAcrossPartitions(), 2);
        assertEquals(configurationData.getConsumerName(), "new-consumer");
        assertEquals(configurationData.getAckTimeoutMillis(), 2);
        assertEquals(configurationData.getTickDurationMillis(), 2);
        assertEquals(configurationData.getPriorityLevel(), 2);
        assertEquals(configurationData.getMaxPendingChunkedMessage(), 2);
        assertTrue(configurationData.isAutoAckOldestChunkedMessageOnQueueFull());
        assertEquals(configurationData.getExpireTimeOfIncompleteChunkedMessageMillis(), 2);
        assertEquals(configurationData.getCryptoFailureAction(), ConsumerCryptoFailureAction.DISCARD);
        assertThat(configurationData.getProperties()).hasSize(1)
            .hasFieldOrPropertyWithValue("new-prop", "new-prop-value");
        assertTrue(configurationData.isReadCompacted());
        assertEquals(configurationData.getSubscriptionInitialPosition(), SubscriptionInitialPosition.Earliest);
        assertEquals(configurationData.getPatternAutoDiscoveryPeriod(), 2);
        assertEquals(configurationData.getRegexSubscriptionMode(), RegexSubscriptionMode.AllTopics);
        assertEquals(configurationData.getDeadLetterPolicy().getDeadLetterTopic(), "new-dlq");
        assertEquals(configurationData.getDeadLetterPolicy().getRetryLetterTopic(), "new-retry");
        assertEquals(configurationData.getDeadLetterPolicy().getInitialSubscriptionName(), "new-dlq-sub");
        assertEquals(configurationData.getDeadLetterPolicy().getMaxRedeliverCount(), 2);
        assertTrue(configurationData.isRetryEnable());
        assertFalse(configurationData.isAutoUpdatePartitions());
        assertEquals(configurationData.getAutoUpdatePartitionsIntervalSeconds(), 2);
        assertTrue(configurationData.isReplicateSubscriptionState());
        assertTrue(configurationData.isResetIncludeHead());
        assertTrue(configurationData.isBatchIndexAckEnabled());
        assertTrue(configurationData.isAckReceiptEnabled());
        assertTrue(configurationData.isPoolMessages());
        assertTrue(configurationData.isStartPaused());
        assertTrue(configurationData.isAutoScaledReceiverQueueSizeEnabled());

        assertNull(configurationData.getMessageListener());
        assertNull(configurationData.getConsumerEventListener());
        assertNull(configurationData.getNegativeAckRedeliveryBackoff());
        assertNull(configurationData.getAckTimeoutRedeliveryBackoff());
        assertNull(configurationData.getMessageListener());
        assertNull(configurationData.getMessageCrypto());
        assertNull(configurationData.getCryptoKeyReader());
        assertNull(configurationData.getBatchReceivePolicy());
        assertNull(configurationData.getKeySharedPolicy());
        assertNull(configurationData.getPayloadProcessor());
    }