[
https://issues.apache.org/jira/browse/KAFKA-966?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13786307#comment-13786307
]
Neha Narkhede commented on KAFKA-966:
-------------------------------------
Jason,
You might want to check out the new proposal for the Consumer API. It allows
you to invoke commitOffsets() on either particular partitions or all partitions
that the consumer is consuming -
https://cwiki.apache.org/confluence/display/KAFKA/Client+Rewrite#ClientRewrite-ConsumerAPI
This should get you some sort of transactionality for message consumption,
where you can invoke commitOffsets() only after your application has processed
the messages successfully. Since commitOffsets() in the new consumer proposal
will use Kafka as the offset storage mechanism, it will be fast and scalable,
unlike Zookeeper offset storage.
> Allow high level consumer to 'nak' a message and force Kafka to close the
> KafkaStream without losing that message
> -----------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-966
> URL: https://issues.apache.org/jira/browse/KAFKA-966
> Project: Kafka
> Issue Type: Improvement
> Components: consumer
> Affects Versions: 0.8
> Reporter: Chris Curtin
> Assignee: Neha Narkhede
> Priority: Minor
>
> Enhancement request.
> The high level consumer is very close to handling a lot of situations a
> 'typical' client would need. Except for when the message received from Kafka
> is valid, but the business logic that wants to consume it has a problem.
> For example if I want to write the value to a MongoDB or Cassandra database
> and the database is not available. I won't know until I go to do the write
> that the database isn't available, but by then it is too late to NOT read the
> message from Kafka. Thus if I call shutdown() to stop reading, that message
> is lost since the offset Kafka writes to ZooKeeper is the next offset.
> Ideally I'd like to be able to tell Kafka: close the KafkaStream but set the
> next offset to read for this partition to this message when I start up again.
> And if there are any messages in the BlockingQueue for other partitions, find
> the lowest # and use it for that partitions offset since I haven't consumed
> them yet.
> Thus I can cleanly shutdown my processing, resolve whatever the issue is and
> restart the process.
> Another idea might be to allow a 'peek' into the next message and if I
> succeed in writing to the database call 'next' to remove it from the queue.
> I understand this won't deal with a 'kill -9' or hard failure of the JVM
> leading to the latest offsets not being written to ZooKeeper but it addresses
> a likely common scenario for consumers. Nor will it add true transactional
> support since the ZK update could fail.
--
This message was sent by Atlassian JIRA
(v6.1#6144)
