Yesterday I came across a weird problem when attempting to run 2 nearly
identical jobs on a cluster. I was able to solve it (or rather workaround
it), but am sharing here so we can consider a potential fix in Flink's
KafkaProducer code.
My scenario is as follows. I have a Flink program that reads from a Kafka
topic, does a simple Map() operation and writes to a Kafka topic with
exactly once semantic. The source and sink topics are configurable, and
the Map() operation is also configurable (i.e. based on CLI arguments it
can choose between a set of Map() operations). The job does not name any
of its operators (this part sounds trivial, right?...but read on). I run
an instance of this program (job) on a Flink standalone cluster running
1.7.1. The cluster has 12 TMs, each with 1 slot each. So basically each
job will run in its own task slot/TM, and hence each job would run in its
own JVM process. The job runs fine, checkpointing regularly and no
errors. However, if I start another instance of the program (with
different source/sink topics), then within a few seconds the first one
fails, and enters recovery. The first one will eventually fail (all
retries exhausted). If I try to start the failed job again, then the
second job would fail within a few seconds. So basically it looked like
one job was tripping over the other. This was especially odd since each
job was running in essentially its own JVM process (i.e. Task Manager /
Task Slot).
Looking at the flink logs, I saw this error message: >> "
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. "
So I looked at the transactionId - and saw that they were of the form: "
transaction.id = Source: Custom Source -> Map -> Sink: Unamed-<guid>",
essentially the transaction.id is set to the description of the chained
operator followed by some GUID. It is not clear to me how the GUID is
generated --- but essentially BOTH my jobs were using the same
transaction.id!
If I understand correctly, Flink uses a pool of KafkaProducers. Each
KafkaProducer within the pool has a transaction.id associated with it. I
think what is happening each of my jobs has its own pool of
KafkaProducers. However, each producer in both pools essentially have the
same ID. So like JobA.Pool: {P1, P2, P3}, JobB.Pool: {P1, P2, P3}.
This sounds like it would not be a problem since each pool will live in
its own JVM process. But since it does break, my conjecture is this --
with the way 2-phase commit works, in the _commit_ phase, I believe the JM
sends a signal to each operator to commit its state. My guess is that
since the IDs collide, the Producer in one pool is told to commit the
transaction with an epoch for a producer in the other pool which happens to
be less than the last epoch for it. Example P1 (in Job A) gets a message
to commit with epoch 0 that is actually meant for P1 (in Job B). The only
other explanation I can think of is that these pools are in fact shared
between task managers -- but that's really hard to believe.
Is my understanding correct?
I was able to solve this by simply naming one of my operators so that the
transaction.id will be unique for each job. Example,
JobA transaction.id = "Source: Custom Source -> (JobA) -> Sink:
unamed-guid"
JobB transaction.id = "Source: Custom Source -> (JobB) -> Sink: unamed-guid"
After I did this - both jobs run successfully.
I think a good improvement would be to _not_ use the job graph description
as the transaction ID. Maybe a simple approach is to require the user to
provide a pool identifier when using Exactly Once with Kafka. At least
this would make it clear.
Thanks
Tim
