@Test
public void testDeserializerOptions() {

    // Reset state on our Test Deserializer
    TestDeserializer.reset();
    // Create a topic with 1 partitions, (partitionId 0)
    final String topicName = "TestTopic" + System.currentTimeMillis();
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(topicName, 1, (short) 1);
    // Create cluster Config
    final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
        .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString())
        .build();
    // Create Deserializer Config
    final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder()
        .withKeyDeserializerClass(TestDeserializer.class)
        .withKeyDeserializerOption("key.option", "key.value")
        .withKeyDeserializerOption("key.option2", "key.value2")
        // Attempt to override a real setting, it should get filtered
        .withKeyDeserializerOption(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "MadeUpValue")
        .withValueDeserializerClass(TestDeserializer.class)
        .withValueDeserializerOption("value.option", "value.value")
        .withValueDeserializerOption("value.option2", "value.value2")
        // Attempt to override a real setting, it should get filtered
        .withValueDeserializerOption(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "MadeUpValue")
        .build();
    // Create Topic Config
    final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName);
    // Create FilterConfig
    final FilterConfig filterConfig = FilterConfig.withNoFilters();
    // Create ClientConfig, instructing to start from tail.
    final ClientConfig clientConfig = ClientConfig.newBuilder()
        .withConsumerId("MyConsumerId")
        .withFilterConfig(filterConfig)
        .withAllPartitions()
        .withStartingPosition(StartingPosition.newTailPosition())
        .withMaxResultsPerPartition(100)
        .withTopicConfig(topicConfig)
        .build();
    // Create consumer
    try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) {
        // We don't actually care to consume anything.  We want to inspect options passed to deserializer.
        final Map<String, Object> passedConfig = TestDeserializer.getConfig();
        // Validate values got passed
        assertEquals("Should have expected value", "key.value", passedConfig.get("key.option"));
        assertEquals("Should have expected value", "key.value2", passedConfig.get("key.option2"));
        assertEquals("Should have expected value", "value.value", passedConfig.get("value.option"));
        assertEquals("Should have expected value", "value.value2", passedConfig.get("value.option2"));
        // Validate didn't overwrite others
        assertNotEquals(
            "Should not have overridden config",
            "MadeUpValue",
            passedConfig.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
        assertNotEquals(
            "Should not have overridden config",
            "MadeUpValue",
            passedConfig.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
    }
    // Reset state
    TestDeserializer.reset();
}