[FLINK-11140] Can not create a Path from an empty string while use BasePathBucketAssigner

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Labels: pull-request-available
• Assignee: Matrix42
• Reporter: Matrix42
• Created: 12/Dec/18 09:00
• Updated: 28/Feb/19 11:15
• Resolved: 09/Jan/19 09:31

Description

while use BasePathBucketAssigner, Flink throw an exception:

Caused by: java.lang.IllegalArgumentException: Can not create a Path from an empty string
at org.apache.flink.core.fs.Path.checkAndTrimPathArg(Path.java:168)
at org.apache.flink.core.fs.Path.<init>(Path.java:181)
at org.apache.flink.core.fs.Path.<init>(Path.java:108)
at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.assembleBucketPath(Buckets.java:309)
at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.getOrCreateBucketForBucketId(Buckets.java:278)
at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onElement(Buckets.java:265)
at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.invoke(StreamingFileSink.java:370)
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
at org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:164)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:94)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)
at java.lang.Thread.run(Thread.java:748)

reason:

BasePathBucketAssigner#getBucketId return an empty string.

@Override
public String getBucketId(T element, BucketAssigner.Context context) {
   return "";
}

while construct a Path, checkAndTrimPathArg methoad will check pathString, if pathString is empty will throw IllegalArgumentException.

public Path(String pathString) {
   pathString = checkAndTrimPathArg(pathString);

  ......
}

Issue Links

• links to: GitHub Pull Request #7287