[BEAM-2979] Fix a race condition in getWatermark() in KafkaIO.

[rangadi]

Two fixes :

• Don't set curRecord to null before updating. If user deserializers throw, ok to keep curRecord pointing to old one.
• Use atomic references for curRecord and curTimestamp since these are accessed in getWatermark() and getCurrentTimestamp(). These methods could be called concurrently with advance().

[coveralls]

Coverage increased (+1.0%) to 69.61% when pulling 70d13ff on rangadi:race_in_kafkaio into 6dd90d8 on apache:master.

[rangadi]

+R @tgroh. Thanks Thomas.

[tgroh]

This is extremely smelly; the current record and current timestamp must be updated atomically and simultaneously, so using two separate atomic references is not sufficient - you need to store the timestamped record here

[rangadi]

Yeah.. I thought of commenting on that explicitly. Right now we don't depend on them to be consistent. But I agree, simpler to do the right thing.

[rangadi]

Fixed. Please take a look.

[coveralls]

Coverage increased (+1.0%) to 69.612% when pulling b4fc45f on rangadi:race_in_kafkaio into 6dd90d8 on apache:master.

[rangadi]

@tgroh PTAL. I changed it to access them within same synchronized block.

[tgroh]

Wait, sorry, I just looked at this again.

A Reader should only be interacted with by a single thread except in very specific circumstances. advance, getCurrent(), getWatermark, and getCurrentTimestamp must all be interacted with in a thread-safe manner (usually this means a single thread). The reader should not have to perform any synchronization within them.

Specifically, the UnboundedReader javadoc states:

  /**
   * A {@code Reader} that reads an unbounded amount of input.
   *
   * <p>A given {@code UnboundedReader} object will only be accessed by a single thread at once.
   */

The synchronization is as such unnecessary

[rangadi]

That was my impression too.. that's we never synchronized around getWatermark(). Good that it is in the contract as well.

Looks like Flink does call getWatermak() concurrently. cc @aljoscha

[rangadi]

@wtanaka, It looks like the runner is not supposed to call getWatermark() concurrently with other calls in the source.
@iemejia @aljoscha, could you comment on this concurrent access in Flink runner? If we need this fix as a short term work around, I could probably merge it with a comment. I don't see much of performance hit.

Source API explicitly mentions single threaded access : https://github.com/apache/beam/blob/release-2.0.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Source.java#L131
getWatermark() documentation does not refer to this explicitly, but might be implicitly expected.

[aljoscha]

Yes, this is a FlinkRunner problem. In UnboundedSourceWrapper the lock scope is not big enough, we synchronise in emitElement() but should instead synchronise inside the reader loop in run().

[rangadi]

Thanks @aljoscha. Will be nice to fix it.
@tgroh : Updated the the fix to remove synchronization. Limited this PR to fixing first issue mentioned in the description: Avoid setting curRecord to null. It should fix the user issue for all practical purposes.

[rangadi]

+R: @chamikaramj, @tgroh

[chamikaramj]

LGTM

[chamikaramj]

Run Java Precommit

[chamikaramj]

Run Java Precommit

[reuvenlax]

tests still failing?

[chamikaramj]

Run Java Precommit

[rangadi]

Run Java Precommit.

[rangadi]

@chamikaramj, how many attempts should we attempt ;-)

[chamikaramj]

Ok, I give up. Will test locally and commit :)