@Test public void testDeserializerOptions() { TestDeserializer.reset(); final String topicName = "TestTopic" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 1, (short) 1); final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder() .withKeyDeserializerClass(TestDeserializer.class) .withKeyDeserializerOption("key.option", "key.value") .withKeyDeserializerOption("key.option2", "key.value2") .withKeyDeserializerOption(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "MadeUpValue") .withValueDeserializerClass(TestDeserializer.class) .withValueDeserializerOption("value.option", "value.value") .withValueDeserializerOption("value.option2", "value.value2") .withValueDeserializerOption(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "MadeUpValue") .build(); final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName); final FilterConfig filterConfig = FilterConfig.withNoFilters(); final ClientConfig clientConfig = ClientConfig.newBuilder() .withConsumerId("MyConsumerId") .withFilterConfig(filterConfig) .withAllPartitions() .withStartingPosition(StartingPosition.newTailPosition()) .withMaxResultsPerPartition(100) .withTopicConfig(topicConfig) .build(); try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) { final Map<String, Object> passedConfig = TestDeserializer.getConfig(); 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")); 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)); } TestDeserializer.reset(); }
public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig) { final KafkaConsumer kafkaConsumer = createConsumer(clientConfig); final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(clientConfig.getTopicConfig().getTopicName()); final Collection<TopicPartition> topicPartitions = new ArrayList<>(); for (final PartitionInfo partitionInfo: partitionInfos) { if (!clientConfig.isPartitionFiltered(partitionInfo.partition())) { topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } } kafkaConsumer.assign(topicPartitions); return kafkaConsumer; }
KafkaConsumerFactory { public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig) { final KafkaConsumer kafkaConsumer = createConsumer(clientConfig); final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(clientConfig.getTopicConfig().getTopicName()); final Collection<TopicPartition> topicPartitions = new ArrayList<>(); for (final PartitionInfo partitionInfo: partitionInfos) { if (!clientConfig.isPartitionFiltered(partitionInfo.partition())) { topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } } kafkaConsumer.assign(topicPartitions); return kafkaConsumer; } }
KafkaConsumerFactory { public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig) { final KafkaConsumer kafkaConsumer = createConsumer(clientConfig); final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(clientConfig.getTopicConfig().getTopicName()); final Collection<TopicPartition> topicPartitions = new ArrayList<>(); for (final PartitionInfo partitionInfo: partitionInfos) { if (!clientConfig.isPartitionFiltered(partitionInfo.partition())) { topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } } kafkaConsumer.assign(topicPartitions); return kafkaConsumer; } KafkaConsumerFactory(final KafkaClientConfigUtil configUtil); }
KafkaConsumerFactory { public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig) { final KafkaConsumer kafkaConsumer = createConsumer(clientConfig); final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(clientConfig.getTopicConfig().getTopicName()); final Collection<TopicPartition> topicPartitions = new ArrayList<>(); for (final PartitionInfo partitionInfo: partitionInfos) { if (!clientConfig.isPartitionFiltered(partitionInfo.partition())) { topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } } kafkaConsumer.assign(topicPartitions); return kafkaConsumer; } KafkaConsumerFactory(final KafkaClientConfigUtil configUtil); KafkaConsumer createConsumer(final ClientConfig clientConfig); KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig); }
KafkaConsumerFactory { public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig) { final KafkaConsumer kafkaConsumer = createConsumer(clientConfig); final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(clientConfig.getTopicConfig().getTopicName()); final Collection<TopicPartition> topicPartitions = new ArrayList<>(); for (final PartitionInfo partitionInfo: partitionInfos) { if (!clientConfig.isPartitionFiltered(partitionInfo.partition())) { topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } } kafkaConsumer.assign(topicPartitions); return kafkaConsumer; } KafkaConsumerFactory(final KafkaClientConfigUtil configUtil); KafkaConsumer createConsumer(final ClientConfig clientConfig); KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig);  }