[jira] [Commented] (KAFKA-19051) Fix implicit acknowledgement cannot be overridden when RecordDeserializationException occurs

["Kuan Po Tseng (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-19051?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17939370#comment-17939370 ] 

Kuan Po Tseng commented on KAFKA-19051:
---------------------------------------

Thanks for reporting this issue [~frouleau], as mentioned in the KIP,

[https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255070434#KIP932:QueuesforKafka-Handlingbadrecords]

There is a statement says:
{quote}If a record was delivered to a consumer but could not be deserialized, the KafkaShareConsumer.poll(Duration) method throws an org.apache.kafka.common.errors.RecordDeserializationException which contains the record's partition and offset information. The KafkaShareConsumer automatically releases the record, although the application can override this if it is using explicit acknowledgement. The next call to KafkaShareConsumer.poll(Duration) skips over the bad record so progress continues to be made.
{quote}
gentle ping [~schofielaj], is this behavior change expected in the implementation? I'm uncertain whether we need to adjust the current implementation or modify the KIP to reflect this change.

> Fix implicit acknowledgement cannot be overridden when RecordDeserializationException occurs
> --------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-19051
>                 URL: https://issues.apache.org/jira/browse/KAFKA-19051
>             Project: Kafka
>          Issue Type: Sub-task
>          Components: consumer
>            Reporter: Frédérik ROULEAU
>            Assignee: Kuan Po Tseng
>            Priority: Major
>
> When a record generates a RecordDeserializationException, KIP mentioned that with explicit acknowledgement the default Release can be overridden.
> When tried, I have:
> {code:java}
> Exception in thread "main" java.lang.IllegalStateException: The record cannot be acknowledged.
>     at org.apache.kafka.clients.consumer.internals.ShareFetch.acknowledge(ShareFetch.java:123)
>     at org.apache.kafka.clients.consumer.internals.ShareConsumerImpl.acknowledge(ShareConsumerImpl.java:683)
>     at org.apache.kafka.clients.consumer.KafkaShareConsumer.acknowledge(KafkaShareConsumer.java:534)
>     at org.example.frouleau.kip932.Main.main(Main.java:62) {code}
> It looks like the record was already released.
> Code used:
> {code:java}
> //....
> } catch (RecordDeserializationException re) {
>     long offset = re.offset();
>     Throwable t = re.getCause();
>     LOGGER.error("Failed to deserialize record at partition={} offset={}", re.topicPartition().partition(), offset, t);
>     ConsumerRecord<String,String> record = new ConsumerRecord<>(re.topicPartition().topic(), re.topicPartition().partition(), offset, "", "");
>     consumer.acknowledge(record, AcknowledgeType.REJECT);
> } {code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)