public static <T> T loadData(Map<String, Object> config,
                                 T existingData,
                                 Class<T> dataCls) {
        try {
            String existingConfigJson = MAPPER.writeValueAsString(existingData);
            Map<String, Object> existingConfig = MAPPER.readValue(existingConfigJson, Map.class);
            Map<String, Object> newConfig = new HashMap<>();
            newConfig.putAll(existingConfig);
            newConfig.putAll(config);
            String configJson = MAPPER.writeValueAsString(newConfig);
            return MAPPER.readValue(configJson, dataCls);
        } catch (IOException e) {
            throw new RuntimeException("Failed to load config into existing configuration data", e);
        }

    }
@Test
    public void testLoadReaderConfigurationData() {
        ReaderConfigurationData confData = new ReaderConfigurationData();
        confData.setTopicName("unknown");
        confData.setReceiverQueueSize(1000000);
        confData.setReaderName("unknown-reader");
        Map<String, Object> config = new HashMap<>();
        config.put("topicNames", ImmutableSet.of("test-topic"));
        config.put("receiverQueueSize", 100);
        confData = ConfigurationDataUtils.loadData(config, confData, ReaderConfigurationData.class);
        assertEquals("test-topic", confData.getTopicName());
        assertEquals(100, confData.getReceiverQueueSize());
        assertEquals("unknown-reader", confData.getReaderName());
    }