o.s.k.r.ReplyingKafkaTemplate : Reply timed out for: ProducerRecord

Question

I am trying to publish some message using kafka consumer with "replyingKafkaTemplate". My main prupose is to subscribe the message, modify it and sending back the modified message. I have tried increasing replyTimeout of replykafkaTemplate. But even that I am getting no response from subscriber. Producer console is showing following.

I have tried increasing transaction timeout, request timeout. But nothing works for me. Any help would be appreciated.

Thanks in advance

These are my config beans :

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> properties = new HashMap<>();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 1000);
    return properties;
}

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> properties = new HashMap<>();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    return properties;
}

 @Bean
public ReplyingKafkaTemplate<String, User, User> replyKafkaTemplate(ProducerFactory<String, User> pf,
        KafkaMessageListenerContainer<String, User> container) {

    ReplyingKafkaTemplate<String, User, User> replyTemplate = new ReplyingKafkaTemplate<>(pf, container);
    replyTemplate.setReplyTimeout(30000);

    return replyTemplate;
}

This is my consumer :

@KafkaListener(topics = "user",containerFactory="kafkaListenerContainerFactory")
@SendTo
public User listen(User user) throws InterruptedException {
    System.out.println("************* message published *************");
    user.setName("myName");
  return user;
}

WARN 8088 --- [TaskScheduler-1] o.s.k.r.ReplyingKafkaTemplate : Reply timed out for: ProducerRecord(topic=user, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = kafka_replyTopic, value = [117, 115, 101, 114]), RecordHeader(key = kafka_correlationId, value = [85, 92, 37, -119, 89, 32, 77, -1, -75, -107, 106, 42, 68, 12, -124, -105]), RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 107, 97, 102, 107, 97, 46, 109, 111, 100, 101, 108, 46, 85, 115, 101, 114])], isReadOnly = true), key=null, value=com.kafka.model.User@71178924, timestamp=null) with correlationId: [113462832283699872744219122180807230615]

Anhlh Anhlh · Accepted Answer · 2019-06-07T03:36:20

You may be need to added sharedReplyTopic in your config ReplyingKafkaTemplate :

@Bean
public ReplyingKafkaTemplate<String, User, User> replyKafkaTemplate(ProducerFactory<String, User> pf,
    KafkaMessageListenerContainer<String, User> container) {
ReplyingKafkaTemplate<String, User, User> replyTemplate = new ReplyingKafkaTemplate<>(pf, container);
replyTemplate.setReplyTimeout(30000);
replyTemplate.setSharedReplyTopic(true);
return replyTemplate;

}

And here is my full example full config:

@Override
@Bean
public Map<String, Object> consumerConfigs() {

Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
props.put(ConsumerConfig.CLIENT_ID_CONFIG, "record-reader");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put("ssl.endpoint.identification.algorithm", ``sslEndpointIdentificationAlgorithm);
props.put("sasl.mechanism", saslMechanism);
props.put("request.timeout.ms", requestTimeoutMs);
props.put("retry.backoff.ms", retryBackoffMs);
props.put("security.protocol", securityProtocol);
props.put("sasl.jaas.config", saslJaasConfig);

return props;
}

@Override
@Bean
public ConsumerFactory<String, String> consumerFactory() {

return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
    new StringDeserializer());
}

@Override
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {

ConcurrentKafkaListenerContainerFactory<String, String> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setReplyTemplate(kafkaTemplate);
return factory;
}

@Override
@Bean
public KafkaMessageListenerContainer<String, String> replyContainer() {
ContainerProperties containerProperties = new ContainerProperties(customerIndexTopic,customerReplyTopic);
return new KafkaMessageListenerContainer<>(consumerFactory(), containerProperties);
 }
 @Override
 @Bean
 public Map<String, Object> producerConfigs() {

Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put("ssl.endpoint.identification.algorithm", sslEndpointIdentificationAlgorithm);
props.put("sasl.mechanism", saslMechanism);
props.put("request.timeout.ms", requestTimeoutMs);
props.put("retry.backoff.ms", retryBackoffMs);
props.put("security.protocol", securityProtocol);
props.put("sasl.jaas.config", saslJaasConfig);

return props;
}

@Override
@Bean
public ProducerFactory<String, String> producerFactory() {

return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Override
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {

return new KafkaTemplate<>(producerFactory());
}

@Override
@Bean
public ReplyingKafkaTemplate<String, String, String> replyKafkaTemplate(
  KafkaMessageListenerContainer<String, String> container) {
ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate = new ReplyingKafkaTemplate<>(producerFactory(),container);
replyingKafkaTemplate.setSharedReplyTopic(true);
replyingKafkaTemplate.setReplyTimeout(10000);
return replyingKafkaTemplate;
}

o.s.k.r.ReplyingKafkaTemplate : Reply timed out for: ProducerRecord

1 Answers