@Test public void testCreatedConsumerValidConfigCanCommunicateToKafka() { String testTopic = "testCreatedConsumer_ValidConfig_CanCommunicateToKafka"; Producer<String, String> testProducer = producerFactory.createProducer(); testProducer.send(new ProducerRecord<>(testTopic, 0, null, null, "foo")); testProducer.flush(); ConsumerFactory<?, ?> testSubject = new DefaultConsumerFactory<>(minimal(kafkaBroker)); testConsumer = testSubject.createConsumer(DEFAULT_GROUP_ID); testConsumer.subscribe(Collections.singleton(testTopic)); assertThat(KafkaTestUtils.getRecords(testConsumer).count()).isOne(); }
@Override public Consumer<K, V> createConsumer(String groupId) { if (this.consumerConfiguration.remove(GROUP_ID_CONFIG) != null) { logger.warn("Found a global {} whilst it is required to be provided consciously", GROUP_ID_CONFIG); } Map<String, Object> consumerConfiguration = new HashMap<>(this.consumerConfiguration); consumerConfiguration.put(GROUP_ID_CONFIG, groupId); return new KafkaConsumer<>(consumerConfiguration); }
DefaultConsumerFactory implements ConsumerFactory<K, V> { @Override public Consumer<K, V> createConsumer(String groupId) { if (this.consumerConfiguration.remove(GROUP_ID_CONFIG) != null) { logger.warn("Found a global {} whilst it is required to be provided consciously", GROUP_ID_CONFIG); } Map<String, Object> consumerConfiguration = new HashMap<>(this.consumerConfiguration); consumerConfiguration.put(GROUP_ID_CONFIG, groupId); return new KafkaConsumer<>(consumerConfiguration); } }
DefaultConsumerFactory implements ConsumerFactory<K, V> { @Override public Consumer<K, V> createConsumer(String groupId) { if (this.consumerConfiguration.remove(GROUP_ID_CONFIG) != null) { logger.warn("Found a global {} whilst it is required to be provided consciously", GROUP_ID_CONFIG); } Map<String, Object> consumerConfiguration = new HashMap<>(this.consumerConfiguration); consumerConfiguration.put(GROUP_ID_CONFIG, groupId); return new KafkaConsumer<>(consumerConfiguration); } DefaultConsumerFactory(Map<String, Object> consumerConfiguration); }
DefaultConsumerFactory implements ConsumerFactory<K, V> { @Override public Consumer<K, V> createConsumer(String groupId) { if (this.consumerConfiguration.remove(GROUP_ID_CONFIG) != null) { logger.warn("Found a global {} whilst it is required to be provided consciously", GROUP_ID_CONFIG); } Map<String, Object> consumerConfiguration = new HashMap<>(this.consumerConfiguration); consumerConfiguration.put(GROUP_ID_CONFIG, groupId); return new KafkaConsumer<>(consumerConfiguration); } DefaultConsumerFactory(Map<String, Object> consumerConfiguration); @Override Consumer<K, V> createConsumer(String groupId); Map<String, Object> configurationProperties(); }
DefaultConsumerFactory implements ConsumerFactory<K, V> { @Override public Consumer<K, V> createConsumer(String groupId) { if (this.consumerConfiguration.remove(GROUP_ID_CONFIG) != null) { logger.warn("Found a global {} whilst it is required to be provided consciously", GROUP_ID_CONFIG); } Map<String, Object> consumerConfiguration = new HashMap<>(this.consumerConfiguration); consumerConfiguration.put(GROUP_ID_CONFIG, groupId); return new KafkaConsumer<>(consumerConfiguration); } DefaultConsumerFactory(Map<String, Object> consumerConfiguration); @Override Consumer<K, V> createConsumer(String groupId); Map<String, Object> configurationProperties();  }