Hi, as I said, I didn't dig too deep into that, but what I saw
was [1].
Generally, if RabbitMQ would have no way to recover subscription
(which I don't think is the case), then it would not be
incompatible with beam, but actually with would be incompatible
any fault tolerant semantics.
[1] https://www.rabbitmq.com/consumers.html#consumer-lifecycle
Dne 14. 11. 2019 13:06 napsal uživatel Daniel Robert
<daniel.rob...@acm.org> <mailto:daniel.rob...@acm.org>:
On 11/14/19 2:32 AM, Jan Lukavský wrote:
Hi Danny,
as Eugene pointed out, there are essentially two "modes
of operation" of CheckpointMark. It can:
a) be used to somehow restore state of a reader (in call
to UnboundedSource#createReader)
b) confirm processed elements in
CheckpointMark#finalizeCheckpoint
If your source doesn't provide a persistent position in
data stream that can be referred to (and serialized -
example of this would be kafka offsets), then what you
actually need to serialize is not the channel, but a way
how to restore it - e.g. by opening a new channel with a
given 'consumer group name'. Then you just use this
checkpoint to commit your processed data in
finalizeCheckpoint.
Note that the finalizeCheckpoint is not guaranteed to be
called - that can happen in cases when an error occurs
and the source has to be rewind back - that is what
direct runner emulates with the probability of
'readerReuseChance'.
I'm reading the documentation of RabbitMQ very quickly,
but if I understand it correctly, then you have to create
a subscription to the broker, serialize identifier of the
subscription into the checkpointmark and then just
recover the subscription in call to
UnboundedSource#createReader. That should do the trick.
I have not seen any such documentation in rabbit. My
understanding is it has to be the same, physical connection
and channel. Can you cite the source you were looking at?
-Danny
Hope this helps, sorry if I'm not using 100% correct
RabbitMQ terminology as I said, I'm not quite familiar
with it.
Best,
Jan
On 11/14/19 5:26 AM, Daniel Robert wrote:
I believe I've nailed down a situation that happens
in practice that causes Beam and Rabbit to be
incompatible. It seems that runners can and do make
assumptions about the serializability (via Coder) of
a CheckpointMark.
To start, these are the semantics of RabbitMQ:
- the client establishes a connection to the server
- client opens a channel on the connection
- messages are either pulled or pushed to the client
from the server along this channel
- when messages are done processing, they are
acknowledged *client-side* and must be acknowledged
on the *same channel* that originally received the
message.
Since a channel (or any open connection) is
non-serializable, it means that a CheckpointMark that
has been serialized cannot ever be used to
acknowledge these messages and correctly 'finalize'
the checkpoint. It also, as previously discussed in
this thread, implies a rabbit Reader cannot accept an
existing CheckpointMark at all; the Reader and the
CheckpointMark must share the same connection to the
rabbit server ("channel").
Next, I've found how DirectRunner (and presumably
others) can attempt to serialize a CheckpointMark
that has not been finalized. In
https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java#L150,
the DirectRunner applies a probability and if it
hits, it sets the current reader to 'null' but
retains the existing CheckpointMark, which it then
attempts to pass to a new reader via a Coder.
This puts the shard, the runner, and the reader with
differing views of the world. In
UnboundedReadEvaluatorFactory's processElement
function, a call to getReader(shard) (
https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java#L132
) clones the shard's checkpoint mark and passes that
to the new reader. The reader ignores it, creating
its own, but even if it accepted it, it would be
accepting a serialized CheckpointMark, which wouldn't
work. Later, the runner calls finishRead (
https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java#L246
). The shard's CheckpointMark (unserialized; which
should still be valid) is finalized. The reader's
CheckpointMark (which may be a different instance)
becomes the return value, which is referred to as
"finishedCheckpoint" in the calling code, which is
misleading at best and problematic at worst as *this*
checkpoint has not been finalized.
So, tl;dr: I cannot find any means of maintaining a
persistent connection to the server for finalizing
checkpoints that is safe across runners. If there's a
guarantee all of the shards are on the same JVM
instance, I could rely on global, static
collections/instances as a workaround, but if other
runners might serialize this across the wire, I'm
stumped. The only workable situation I can think of
right now is to proactively acknowledge messages as
they are received and effectively no-op in
finalizeCheckpoint. This is very different,
semantically, and can lead to dropped messages if a
pipeline doesn't finish processing the given message.
Any help would be much appreciated.
Thanks,
-Danny
On 11/7/19 10:27 PM, Eugene Kirpichov wrote:
Hi Daniel,
This is probably insufficiently well documented.
The CheckpointMark is used for two purposes:
1) To persistently store some notion of how much
of the stream has been consumed, so that if
something fails we can tell the underlying
streaming system where to start reading when we
re-create the reader. This is why CheckpointMark
is Serializable. E.g. this makes sense for Kafka.
2) To do acks - to let the underlying streaming
system know that the Beam pipeline will never
need data up to this CheckpointMark. Acking does
not require serializability - runners call ack()
on the same in-memory instance of CheckpointMark
that was produced by the reader. E.g. this makes
sense for RabbitMq or Pubsub.
In practice, these two capabilities tend to be
mutually exclusive: some streaming systems can
provide a serializable CheckpointMark, some can
do acks, some can do neither - but very few (or
none) can do both, and it's debatable whether it
even makes sense for a system to provide both
capabilities: usually acking is an implicit form
of streaming-system-side checkpointing, i.e. when
you re-create the reader you don't actually need
to carry over any information from an old
CheckpointMark - the necessary state (which
records should be delivered) is maintained on the
streaming system side.
These two are lumped together into one API simply
because that was the best design option we came
up with (not for lack of trying, but suggestions
very much welcome - AFAIK nobody is happy with it).
RabbitMQ is under #2 - it can't do serializable
checkpoint marks, but it can do acks. So you can
simply ignore the non-serializability.
On Thu, Nov 7, 2019 at 12:07 PM Daniel Robert
<daniel.rob...@acm.org
<mailto:daniel.rob...@acm.org>> wrote:
(Background: I recently upgraded RabbitMqIO
from the 4.x to 5.x library.
As part of this I switched to a pull-based
API rather than the
previously-used push-based. This has caused
some nebulous problems so
put up a correction PR that I think needs
some eyes fairly quickly as
I'd consider master to be broken for rabbitmq
right now. The PR keeps
the upgrade but reverts to the same
push-based implementation as in 4.x:
https://github.com/apache/beam/pull/9977 )
Regardless, in trying to get the pull-based
API to work, I'm finding the
interactions between rabbitmq and beam with
CheckpointMark to be
fundamentally impossible to implement so I'm
hoping for some input here.
CheckointMark itself must be Serializable,
presumably this means it gets
shuffled around between nodes. However
'Channel', the tunnel through
which it communicates with Rabbit to ack
messages and finalize the
checkpoint, is non-Serializable. Like most
other CheckpointMark
implementations, Channel is 'transient'. When
a new CheckpointMark is
instantiated, it's given a Channel. If an
existing one is supplied to
the Reader's constructor (part of the
'startReader()' interface), the
channel is overwritten.
*However*, Rabbit does not support 'ack'ing
messages on a channel other
than the one that consumed them in the first
place. Attempting to do so
results in a '406 (PRECONDITION-FAILED) -
unknown delivery tag'. (See
https://www.grzegorowski.com/rabbitmq-406-channel-closed-precondition-failed
).
Truthfully, I don't really understand how the
current implementation is
working; it seems like a happy accident. But
I'm curious if someone
could help me debug and implement how to
bridge the
re-usable/serializable CheckpointMark
requirement in Beam with this
limitation of Rabbit.
Thanks,
-Daniel Robert
