[SPARK-17841][STREAMING][KAFKA] drain commitQueue

[koeninger]

What changes were proposed in this pull request?

Actually drain commit queue rather than just iterating it.
iterator() on a concurrent linked queue won't remove items from the queue, poll() will.

How was this patch tested?

Unit tests

[SparkQA]

Test build #66609 has finished for PR 15407 at commit 8810a79.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rxin]

Is there a test we can add to cover this?

I realized "How was this patch tested?" was confusing. That was meant to ask how is the change tested itself, and a bug fix obviously can't be tested by existing unit tests.

[koeninger]

This doesn't affect correctness (only the highest offset for a given partition is used in any case), just memory leaks. I'm not sure what a good way to unit test memory leaks is, short of exposing the internal state of that queue, which seems less than optimal.

[tdas]

Can you explain what is the memory that can currently leak with the iterator?

[rxin]

@koeninger can you put more information into the description of the pull request? At the very least we should talk about the current implementation causes memory leaks.

[koeninger]

@rxin @tdas right now, items to be committed can be added to the queue, but they will never actually be removed from the queue. poll() removes, iterator() does not. I updated the description of the PR.

[rxin]

Thanks for the explanation. Merging in master/branch-2.0.

[zsxwing]

LGTM