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);  }
@Test public void testBasicConsumerMultiplePartitions() { final int maxRecordsPerPoll = 10; final String topicName = "TestTopic"; sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 0); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 1); final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder() .withKeyDeserializerClass(StringDeserializer.class) .withValueDeserializerClass(StringDeserializer.class) .build(); final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName); final ClientConfig clientConfig = ClientConfig.newBuilder() .withConsumerId("MyConsumerId") .withNoFilters() .withAllPartitions() .withMaxResultsPerPartition(maxRecordsPerPoll) .withTopicConfig(topicConfig) .build(); try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2)); assertEquals("Should have found " + maxRecordsPerPoll + " records", maxRecordsPerPoll, records.count()); records = consumer.poll(Duration.ofSeconds(2)); assertEquals("Should have found " + maxRecordsPerPoll + " records", maxRecordsPerPoll, records.count()); } }
@Test public void testBasicConsumerExcludePartitions() { final int maxRecordsPerPoll = 10; final String topicName = "TestTopic"; sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 0); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 1); final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder() .withKeyDeserializerClass(StringDeserializer.class) .withValueDeserializerClass(StringDeserializer.class) .build(); final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName); final ClientConfig clientConfig = ClientConfig.newBuilder() .withConsumerId("MyConsumerId") .withNoFilters() .withPartition(1) .withMaxResultsPerPartition(maxRecordsPerPoll) .withTopicConfig(topicConfig) .build(); try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2)); assertEquals("Should have found " + maxRecordsPerPoll + " records", maxRecordsPerPoll, records.count()); for (final ConsumerRecord<String, String> record: records) { assertEquals("Should be from parittion 1 only", 1, record.partition()); } records = consumer.poll(Duration.ofSeconds(2)); assertTrue("Should be empty", records.isEmpty()); } }
@Test public void testBasicConsumerWithRecordFilter() { final int maxRecordsPerPoll = 10; final String topicName = "TestTopic"; sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 0); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(maxRecordsPerPoll, topicName, 1); final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder() .withKeyDeserializerClass(StringDeserializer.class) .withValueDeserializerClass(StringDeserializer.class) .build(); final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName); final FilterConfig filterConfig = FilterConfig.withFilters( new RecordFilterDefinition(new PartitionFilter(), new HashMap<>()) ); final ClientConfig clientConfig = ClientConfig.newBuilder() .withConsumerId("MyConsumerId") .withFilterConfig(filterConfig) .withAllPartitions() .withMaxResultsPerPartition(maxRecordsPerPoll * 100) .withTopicConfig(topicConfig) .build(); try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10)); assertEquals("Should have found " + maxRecordsPerPoll + " records", maxRecordsPerPoll, records.count()); for (final ConsumerRecord<String, String> record: records) { assertEquals("Should be from parittion 1 only", 1, record.partition()); } records = consumer.poll(Duration.ofSeconds(2)); assertTrue("Should be empty", records.isEmpty()); } }
@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(); }