Hi,
I'm looking for a way to make a sync commit of offsets to kafka 0.10?
commitAsync works well, but I'd like to proceed to next job only after
successful commit, a small additional latency is not an issue for my
usecase. I know I can store offsets somewhere else, but builtin kafka offset
storage looks good and easy to use. What is the correct way to do this?
I tried first approach that came up to my mind, like this:
val latch = new
commitAsync(offsetRanges, new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition,
OffsetAndMetadata> offsets, Exception exception) {
latch.countDown();
}
});
latch.await();
but this does not work - I get very wierd effects - new data from kafka is
read with great delays.
Is there an elegant right way to do this?
--
Sent from: http://apache-spark-user-list.1001560.n3.nabble.com/
---------------------------------------------------------------------
To unsubscribe e-mail: [email protected]
