You must use get() to confirm delivery, if that is critical, but you should use the variant with a timeout, otherwise you risk hanging indefinitely.
To clarify, though; "exactly once" really only applies to kafkaRead/process/KafkaWrite and then the only guarantee is that the write and read offset commit are atomic; the process step is at least once.
http/process/kafkaWrite is simply transactional.
If it's simply write to kafka and there is no other possibility of failure thereafter then, no, you probably don't need transactions. If the POST might fail after the send, then you might want to use transactions (and read_committed on the consumer).
I am not sure what you mean by "idempotent producer"; idempotency usually applies to consumers (i.e. they can handle
duplicate messages).
if (kafkaTemplate.isInTransaction()) kafkaTemplateWithTx.send() else kafkaTemplateWithoutTx.send()
I am trying to implement Producer and Consumer in different modules with java + spring boot and Avro schema.
I am producing messages successfully, there is 200 status on schema registry server when the producer sends message with avro schema, but i have errors when this message arrives at the consumer. Looks like consumer doesn't know nothing about the schema registry server and schema format at all. There is no request to the schema registry server to get the schema format.
Is there any example with producer, consumer in different modules ?
consumer:
...
value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
properties:
schema.registry.url: "localhost:8081 - or another url to your schema registry"@filip-halemba
This is my configuration for the consumer
@Configuration
public class KafkaConsumerConfiguration
{
@Value(value = "${spring.kafka.bootstrapAddress}")
private String bootstrapAddress;
@Value(value = "${spring.kafka.schemaRegistryAddress}")
private String schemaRegistryAddress;
@Bean
public Map<String, Object> consumerConfigs()
{
Map<String, Object> props = new HashMap<>();
// list of host:port pairs used for establishing the initial connections to the Kafka cluster
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "helloworld");
return props;
}
@Bean
public ConsumerFactory<String, Object> consumerFactory()
{
return new DefaultKafkaConsumerFactory(consumerConfigs(), new KafkaAvroDeserializer(), new KafkaAvroDeserializer());
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> kafkaListenerContainerFactory()
{
ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}And I have another class which is Consumer class :
@KafkaListener(topics = "new_audit_topic_avro", group_id="helloworld")
public void consumeMessages(ConsumerRecord<String, GenericRecord> record) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(10);
logger.info("received payload='{}'", record.toString());
latch.await();
}
MessageHandlerMethodFactory. It might be tricky to provide an extension point to add argument resolvers, because the order might be important. Feel free to open a new feature issue in GitHub (and contributions are welcome).
@filip-halemba Here is the error message :
2020-01-21 15:10:27 - Commit list: {}
2020-01-21 15:10:27 - Error while processing: null
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition new_audit_topic_avro-0 at offset 0. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 21
Caused by: java.lang.NullPointerException: null
at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:122) ~[kafka-avro-serializer-3.3.1.jar:na]
at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:93) ~[kafka-avro-serializer-3.3.1.jar:na]
at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55) ~[kafka-avro-serializer-3.3.1.jar:na]
at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1268) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher.access$3600(Fetcher.java:124) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1492) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1600(Fetcher.java:1332) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:645) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:606) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1263) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1225) ~[kafka-clients-2.3.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1201) ~[kafka-clients-2.3.1.jar:na]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:993) ~[spring-kafka-2.3.4.RELEASE.jar:2.3.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:949) ~[spring-kafka-2.3.4.RELEASE.jar:2.3.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:901) ~[spring-kafka-2.3.4.RELEASE.jar:2.3.4.RELEASE]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_222-1-ojdkbuild]
at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) [na:1.8.0_222-1-ojdkbuild]
at java.util.concurrent.FutureTask.run(FutureTask.java) [na:1.8.0_222-1-ojdkbuild]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_222-1-ojdkbuild]
I have Consumer in one module, and I need to receive message with Avro schema serializes.
I haven't information about the schema (it's defined in some other module).
How to create my Consumer? What kind of parameter should I receive ?
How this module will communicate with schema registry service ?
@KafkaListener(topics = "new_audit_topic_avro", groupId = "helloworld")
public void consumeMessages(ConsumerRecord<String, GenericRecord> record) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
logger.info("received payload='{}'", record.toString());
latch.await();
}
This is not working.
Please learn how to format code (see the Markdown link to the right). Caused by: java.lang.NullPointerException: null this is in the deserializer itself and has nothing to do with Spring - I can't tell what the NPE is caused by because the line numbers don't line up with the source version I have. Use a debugger to see why you are getting an NPE - it's either a problem on the producer side or the schema can't be found in the registry.
You should ask general questions about Kafka/Avro to the wider Kafka community; we are not experts on Avro here.
Hi! I have a (transactional) @KafkaListener that might take a long time (multiple minutes) to handle a single message (which is fine in itself, it does a bunch of remote calls to external APIs that is also a bit flaky and slow). When the listener then finishes and tries to commit the offset it fails with org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.
To my understanding the reason for this is the transaction.timeout.ms which defaults to 60 seconds. Is that correct? Can I increase this limit somehow for this specific listener? I already have an own ConcurrentKafkaListenerContainerFactory for it. But there I can only set consumer properties. However transaction.timeout.ms is a producer setting. Where does the producer that commit the offset come from and how can I change the setting for it (without setting it globally for all producers)?
SeekToCurrentErrorHandler on my ListenerContainerFactory. If I understood you correctly (and the code of SeekToCurrentErrorHandler at which I just looked) the messages already get ack-ed immediatly when I use it?
unless you are using MANUAL_IMMEDIATE ack mode
Is this maybe what I want (stupid question, I know)? If I read this it will re-send the record to the same topic-partition when MANUAL_IMMEDIATE is true? So I could configure my AckMode to MANUAL_IMMEDIATE and then .acknowledge() at the start of my listener (which would commit the offset?). And one Exception the SeekToCurrentErrorHandler will re-send the record for me.
There is then a slight chance that I could loose a message if my service is killed after calling ack but that would be okay for me.
Okay, but max.poll.interval.ms is a consumer setting which I could just change for the ConsumerFactory in my ListenerContainerFactory.
It's not at all clear why you want to commit the offset so early.
I assume it's because I don't an in-depth understanding what my problem is :/ Or maybe I have stated my problem not clearly enough:
I have a listener that takes it's sweet time. I already only poll on record but I still get a ProducerFencedException when my listener finally finishes and commits. So I assume committing the offset early would solve my problem "neatly".
@Transactional(transactionManager = "kafkaTransactionManager") in the first place. The only thing I do i take the incoming data, process it and then write to the database. And all the database interaction is already behind a JPA transaction manager to it will be rolled back anyway in case of an uncaught exception. I don't have any other interaction with Kafka other than consuming.
ChainedKafkaTransactionManager into the listener container; when using @Transactional, the container knows nothing about the Kafka transaction and cannot send the consumed offset(s) to the transaction (which is how Kafka makes consumption and production atomic).