Matrix42 created FLINK-11140:
--------------------------------
Summary: Can not create a Path from an empty string while use BasePathBucketAssigner
Key: FLINK-11140
URL:
https://issues.apache.org/jira/browse/FLINK-11140 Project: Flink
Issue Type: Bug
Components: Streaming
Reporter: Matrix42
Assignee: Matrix42
while use BasePathBucketAssigner, Flink throw an exception:
{code:java}
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)
{code}
reason:
BasePathBucketAssigner#getBucketId return an empty string.
{code:java}
@Override
public String getBucketId(T element, BucketAssigner.Context context) {
return "";
}
{code}
while construct a Path, checkAndTrimPathArg methoad will check pathString, if pathString is empty will throw IllegalArgumentException.
{code:java}
public Path(String pathString) {
pathString = checkAndTrimPathArg(pathString);
......
}{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
Locked
