ExecutionException instead of UnknownTopicOrPartitionException

[Jean-Charles Jabouille <je...@kelkoo.com>]

Hi,

I'm currently developing an application to use Kafka in Java. My application just push an offer synchronously in a topic. I have 3 brokers and 3 zookeeper instance. I want to catch Exception in order my process does not crash but try to retry and do some code for specific exception.

So I run my application then crash Kafka brokers. The process is well blocked when trying to push a offer in the topic. When I restart my 3 kafka servers my application crash with a "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition". The exception is an ExecutionException and the causeBy of this exception is an UnknownTopicOrPartitionException.

What I don't understand is that in my application I have a "catch (RetriableException e)". And From my understanding, "UnknownTopicOrPartitionException" extends "RetriableException". But the exception is never catch as it's an ExecutionException that i So Why Kafka throw a "ExecutionException" instead of a "UnknownTopicOrPartitionException" is throwed.

The stack:
"Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:56)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:43)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:25)"


Thanks,

jc

Kelkoo SAS
Société par Actions Simplifiée
Au capital de € 4.168.964,30
Siège social : 158 Ter Rue du Temple 75003 Paris
425 093 069 RCS Paris

Ce message et les pièces jointes sont confidentiels et établis à l'attention exclusive de leurs destinataires. Si vous n'êtes pas le destinataire de ce message, merci de le détruire et d'en avertir l'expéditeur.

Re: ExecutionException instead of UnknownTopicOrPartitionException

[Guozhang Wang <wa...@gmail.com>]

Hi Jean-Charles,

The Future<RecordMetadata>.get() will always throw an
ExecutionException(Throwable: CauseException), so in your code should look
like sth.:

-----------------------

try {
  future = producer.send(..);
} catch (KafkaException e) {
  // handle any KafkaException that is not ApiException.
}

try {
  future.get();
} catch {ExecutionException e) {
  ApiException cause = e.getCause();
  // handle the cause exception
}

-----------------------

Guozhang

On Wed, Jul 15, 2015 at 1:27 AM, Jean-Charles Jabouille <
jean-charles.jabouille@kelkoo.com> wrote:

>
> Hi,
>
> I'm currently developing an application to use Kafka in Java. My
> application just push an offer synchronously in a topic. I have 3 brokers
> and 3 zookeeper instance. I want to catch Exception in order my process
> does not crash but try to retry and do some code for specific exception.
>
> So I run my application then crash Kafka brokers. The process is well
> blocked when trying to push a offer in the topic. When I restart my 3 kafka
> servers my application crash with a
> "java.util.concurrent.ExecutionException:
> org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This
> server does not host this topic-partition". The exception is an
> ExecutionException and the causeBy of this exception is an
> UnknownTopicOrPartitionException.
>
> What I don't understand is that in my application I have a "catch
> (RetriableException e)". And From my understanding,
> "UnknownTopicOrPartitionException" extends "RetriableException". But the
> exception is never catch as it's an ExecutionException that i So Why Kafka
> throw a "ExecutionException" instead of a
> "UnknownTopicOrPartitionException" is throwed.
>
> The stack:
> "Caused by: java.util.concurrent.ExecutionException:
> org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This
> server does not host this topic-partition.
>     at
> org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:56)
>     at
> org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:43)
>     at
> org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:25)"
>
>
> Thanks,
>
> jc
>
> Kelkoo SAS
> Société par Actions Simplifiée
> Au capital de € 4.168.964,30
> Siège social : 158 Ter Rue du Temple 75003 Paris
> 425 093 069 RCS Paris
>
> Ce message et les pièces jointes sont confidentiels et établis à
> l'attention exclusive de leurs destinataires. Si vous n'êtes pas le
> destinataire de ce message, merci de le détruire et d'en avertir
> l'expéditeur.
>



-- 
-- Guozhang