[FLINK-11140][streaming] Fix empty child path check in Buckets

[Matrix42]

What is the purpose of the change

Fix empty child path check in Buckets

Brief change log

add check for empty child path in Buckets

Verifying this change

BucketsTest#testAssembleBucketPath()

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): ( no)
• The public API, i.e., is any changed class annotated with @public(Evolving): (no)
• The serializers: ( no)
• The runtime per-record code paths (performance sensitive): (no )
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no )
• The S3 file system connector: (no )

Documentation

• Does this pull request introduce a new feature? ( no)
• If yes, how is the feature documented? (not documented)

[yanghua]

Thanks for your contribution @Matrix42 . I left one minor suggestion. In addition, It would be better to add a test for this fix. What's more, please follow the PR template and fill the section of Brief change log and Verifying this change out correctly.

[yanghua]

Now that you have given bucketId.toString() to the variable child, you can change it to: return new Path(basePath, child);

[Matrix42]

@yanghua thanks for your review, i will fix this

[zentol]

IMO an empty bucket Id doesn't make any sense as it effectively disables bucketing.
As far as i can tell this is an implementation error on the user-side requiring no change on our side.
@kl0u WDYT?

[zentol]

Oh for heaven's sake this is one of our classes isn't it. god damn it.

[aljoscha]

@zentol I think this is a valid fix because it's our class, right?

@Matrix42 Can you please reopen the PR?

[kl0u]

@Matrix42 I think this is a valid issue. I will have a look to the PR now and thanks for reporting it.

[kl0u]

@Matrix42 I have reviewed your PR and I have some comments. Given that the PR is closed and I cannot submit a review, I pushed my changes in my branch here: https://github.com/kl0u/flink/tree/buckets

Feel free to integrate them and open a new PR and I can review it again.

In a nutshell, the changes that I propose are:

1. move the new test in a separate class called BucketAssignerITCases where we can add in the future more tests
2. now the test just checks that no exception is thrown, but it does not test that the path is the expected one. For this, I propose to change the return type of the Buckets#onElement(...) to Bucket<IN, BucketID>, and do assertions on that.

All these are included in my branch. Feel free to ping me when you open the new PR.

[Matrix42]

@zentol @kl0u thanks for your review, I will reopen this PR.

[kl0u]

Thanks @Matrix42 ! Please ping me when you integrate the comments.

[kl0u]

Sorry, I just saw that you addressed the comments. I will review soon.

[kl0u]

Thanks for reporting the issue and working on it @Matrix42! Changes look good. I will merge as soon as Travis gives the green light.