[BEAM-4347] Enforce ErrorProne analysis in kafka IO by timrobertson100 · Pull Request #5422 · apache/beam

timrobertson100 · 2018-05-18T13:58:29Z

A simple PR to enforce error prone and fail the build on warnings raised. Current errorprone warnings are fixed (but I did not do an IDEA code analysis in this editing round).

CC @iemejia @swegner - could you please assign a reviewer?

iemejia · 2018-05-18T14:06:55Z

R: @rangadi Can you please take a look at this one since it is Kafka related.

rangadi

Thanks for doing this @timrobertson100. Left a few comments (TODO comments and checking if switch to 'ArrayList' is required)

@iemejia, LGTM overall.

rangadi · 2018-05-18T17:51:24Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java

        ProducerSpEL.beginTransaction(producer);
      }

+      @SuppressWarnings("FutureReturnValueIgnored")


Please leave a TODO for me.
// TODO : rangadi : add explanation for why this is ok.
Can we make this send() return future and ignore it at the call site?

In general I'm not much in favour of leaving TODOs within code (looks cluttered, doesn't help with tracking outstanding work etc).
How about I open a jira task to review future handling in KafkaIO, informing that we've suppressed warnings (doesn't change current behaviour) but would like to review if we may drop messages on exceptional cases? I'll be happy to dig deeper with you on that.

Makes sense. Please file a jira. I wanted to add the comment soon after this gets merged. I know why it is ok to ignore here (it will be waited on inside commitTxn()). You can add the explanation in PR itself. If you are adding the comment, I suggest changing sendRecord() return the future and let the call ignore it with a comment saying they are flushed in commitTxn().

I'll make the code change - new commit later

rangadi · 2018-05-18T17:55:58Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java

     * closed if it is stays in cache for more than 1 minute, i.e. not used inside
     * KafkaExactlyOnceSink DoFn for a minute.
     */
+    @SuppressWarnings("FutureReturnValueIgnored")


I don't see where a Future is ignored. Please leave a comment :
// TODO : rangadi : review if this is even required.

FYI: It's needed because of the SCHEDULED_CLEAN_UP_THREAD.scheduleAtFixedRate(...)

Please add a comment to that effect. Will be useful for other readers.

rangadi · 2018-05-18T18:12:26Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java

  }

  @ProcessElement
+  @SuppressWarnings("FutureReturnValueIgnored")


// TODO : rangadi : explain why this is ok.

As above - can I include this in a Review future handling in KafkaIO Jira instead please?

Yes, either jira, or better, you can add a comment. Reason it is ok to ignore here is that SendCallback() keeps track of failures, which are checked inside finishBundle().

rangadi · 2018-05-18T18:14:35Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java


    // cycle through the partitions in order to interleave records from each.
-    curBatch = Iterators.cycle(new LinkedList<>(partitionStates));
+    curBatch = Iterators.cycle(new ArrayList<>(partitionStates));


Was this also recommended by ErrorProne?

Yes.

LinkedList almost never out-performs ArrayList or ArrayDeque. If you are using LinkedList as a list, prefer ArrayList. If you are using LinkedList as a stack or queue/deque, prefer ArrayDeque.

Because ArrayDeque rejects nulls I erred on the side of caution

We are using LinkedList because of elements deleted in random order (in advance()). There are no nulls. Was it an ErrorProne error?
Overall I don't think the difference matters at all either way. I would be surprised if ErrorProne insisted we change it.

Error prone reports it as warning but we now fail builds on warning (unless suppressed of course). Thanks for confirming no nulls.

rangadi · 2018-05-18T18:15:58Z

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java

                    (record -> new Instant(TimeUnit.SECONDS.toMillis(record.getKV().getValue())
                                             + customTimestampStartMillis)),
-                   Duration.millis(0),
+                   Duration.ZERO,


Just curious: ErrorProne suggested this?

Yes.
It offers some nice suggestions for readability I've found (e.g. seconds(180) -> minutes(3))

timrobertson100 · 2018-05-19T05:42:04Z

Thanks for the review @rangadi

If you're ok with the proposal I create a Jira for a review of Future handling in KafkaIO I don't think there are further changes. Alternatively, we can do that exploration now - if we find changes are needed they may be more invasive and warrant a specific jira for that anyway though.

rangadi · 2018-05-20T01:54:38Z

Commented. Since you are interested, simpler thing might be to add the explanations for ignoring futures in this PR itself. JIRA is also fine.

timrobertson100 · 2018-05-20T15:10:17Z

Thanks @rangadi. I'll make a few changes and provide another commit (probably tomorrow or Tuesday)

timrobertson100 · 2018-05-22T10:24:13Z

Please can you take a look @rangadi ?

rangadi

LGTM. Suggested two minor changes : removing 'Note:' in comments and reverting ArrayDeque change.

rangadi · 2018-05-23T16:21:11Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java

      KafkaExactlyOnceSink.ensureEOSSupport();
    }

+    // Note: Futures ignored as exceptions will be flushed out in the commitTxn


A short comment is already a note :). We could remove them.

rangadi · 2018-05-23T16:25:31Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java


    // cycle through the partitions in order to interleave records from each.
-    curBatch = Iterators.cycle(new ArrayList<>(partitionStates));
+    curBatch = Iterators.cycle(new ArrayDeque<>(partitionStates));


This list is neither used as a stack or nor as a queue. I would say you can either leave it as ArrayList or LinkedList. ArrayDeque incorrectly implies usage.

timrobertson100 · 2018-05-23T18:05:48Z

Thanks for review @rangadi - those changes now applied.

rangadi · 2018-05-23T18:35:10Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java

  }

-  // Note: Suppression since rrrors are tracked in SendCallback(), and checked on finishBundle()
+  // Suppression since rrrors are tracked in SendCallback(), and checked on finishBundle()


typo in 'errors'. Also 'in finishBundle()'?.

rangadi · 2018-05-23T18:37:11Z

LGTM. Thanks @timrobertson100.
@iemejia : ready to merge (couple of typos are optional, which might already be fixed by the time you merge).

iemejia · 2018-05-23T20:30:40Z

Oups I screwed the correct title by double clicking merge by mistake. Thanks @timrobertson100 and @rangadi for the proper review.

iemejia · 2018-05-23T20:31:31Z

LGTM in postfix mode due to my sloppy fingers.

[BEAM-4347] Enforce ErrorProne analysis in kafka IO

41f75b1

iemejia self-requested a review May 18, 2018 14:07

rangadi suggested changes May 18, 2018

View reviewed changes

[BEAM-4347] Improve ErrorProne analysis in kafka IO

aa25105

rangadi suggested changes May 23, 2018

View reviewed changes

[BEAM-4347] Minor changes following PR review

2750999

rangadi reviewed May 23, 2018

View reviewed changes

[BEAM-4347] Correction of typo following PR review

585547a

iemejia merged commit e31f27f into apache:master May 23, 2018

Conversation

timrobertson100 commented May 18, 2018

Uh oh!

iemejia commented May 18, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rangadi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

timrobertson100 May 19, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rangadi May 18, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

timrobertson100 May 19, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

timrobertson100 commented May 19, 2018

Uh oh!

rangadi commented May 20, 2018

Uh oh!

timrobertson100 commented May 20, 2018

Uh oh!

timrobertson100 commented May 22, 2018

Uh oh!

rangadi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

timrobertson100 commented May 23, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rangadi commented May 23, 2018

Uh oh!

iemejia commented May 23, 2018

Uh oh!

iemejia commented May 23, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

iemejia commented May 18, 2018 •

edited

Loading

timrobertson100 May 19, 2018 •

edited

Loading

rangadi May 18, 2018 •

edited

Loading

timrobertson100 May 19, 2018 •

edited

Loading