@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()); } }
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);  }