4

My spring/java consumer is not able to access the message produced by producer. However, when I run the consumer from console/terminal it is able to receive the message produced by spring/java producer.

Consumer Configuration :

@Component
@ConfigurationProperties(prefix="kafka.consumer")
public class KafkaConsumerProperties {

    private String bootstrap;
    private String group;
    private String topic;

    public String getBootstrap() {
        return bootstrap;
    }

    public void setBootstrap(String bootstrap) {
        this.bootstrap = bootstrap;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }
}

Listener Configuration :

@Configuration
@EnableKafka
public class KafkaListenerConfig {

    @Autowired
    private KafkaConsumerProperties kafkaConsumerProperties;

    @Bean
    public Map<String, Object> getConsumerProperties() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperties.getBootstrap());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroup());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        return properties;
    }

    @Bean
    public Deserializer stringKeyDeserializer() {
        return new StringDeserializer();
    }

    @Bean
    public Deserializer transactionJsonValueDeserializer() {
        return new JsonDeserializer(Transaction.class);
    }

    @Bean
    public ConsumerFactory<String, Transaction> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(getConsumerProperties(), stringKeyDeserializer(), transactionJsonValueDeserializer());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Transaction> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Transaction> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConcurrency(1);
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

Kafka Listener :

@Service
public class TransactionConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(Transaction.class);

    @KafkaListener(topics={"transactions"}, containerFactory = "kafkaListenerContainerFactory")
    public void onReceive(Transaction transaction) {
        LOGGER.info("transaction = {}",transaction);
    }
}

Consumer Application :

@SpringBootApplication
public class ConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }

}

TEST CASE 1 : PASS I started my spring/java producer and run the consumer from console. When I produce message form producer my console consumer is able to access the message.

TEST CASE 2 : FAILED I started my spring/java consumer and run the producer from console. When I produce message form console producer my spring/java consumer is not able to access the message.

TEST CASE 3 : FAILED I started my spring/java consumer and run the spring/java producer. When I produce message form spring/java producer my spring/java consumer is not able to access the message.

Question

  1. Is there anything wrong in my consumer code ?

  2. Am I missing any configuration for my kafka-listener?

  3. Do I need to explicitly run the listener? (I don't think so since I can see on the terminal log connecting to topic, still I am not sure)

1
  • Could it be related to the way your testing it? Did you find the solution?
    – Sylhare
    Jul 9, 2020 at 20:05

1 Answer 1

0

Okay you are missing AUTO_OFFSET_RESET_CONFIG in Consumer Configs

properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

auto.offset.reset

What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted):

earliest: automatically reset the offset to the earliest offset

latest: automatically reset the offset to the latest offset

none: throw exception to the consumer if no previous offset is found for the consumer's group

anything else: throw exception to the consumer

Note : auto.offset.reset to earliest will work only if kafka does not have offset for that consumer group (So in you case need to add this property with new consumer group and restart the application)

4
  • Thank you for the reply. I tried the solution that you proposed, however, it did not work. I tried with all the option but no luck. Do you know how to set '--from-beginning' option (that we use in console) in Java code ?
    – Developer
    Apr 11, 2019 at 13:56
  • 1
    Do you know how to set '--from-beginning' option (that we use in console) in Java code ? that is in my answer @Developer
    – Ryuzaki L
    Apr 11, 2019 at 14:56
  • I have exactly the same problem scenario. This solution have no luck to me either. Did you find any possible work around?
    – Alfran
    Jun 24, 2020 at 16:35
  • Try spring.kafka.consumer.auto-offset-reset=earlisest (see docs)
    – Tim Büthe
    Sep 27, 2022 at 20:22

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.