can't read/write hdfs in Flink CLUSTER(Standalone)

i just write a simple demo like:

```

Configuration conf = new Configuration();
conf.set("fs.default.name", "hdfs://localhost:9000");
//other
codes
p.apply("ReadLines", TextIO.read().from("hdfs://localhost:9000/tmp/words"))

.apply(TextIO.write().to("hdfs://localhost:9000/tmp/hdfsout"));

```

it works in flink local model with cmd:

```

mvn exec:java -Dexec.mainClass=com.joe.FlinkWithHDFS -Pflink-runner -Dexec.args="--runner=FlinkRunner
--filesToStage=target/flinkBeam-2.2.0-SNAPSHOT-shaded.jar"

```

but not works in CLUSTER mode:

```

mvn exec:java -Dexec.mainClass=com.joe.FlinkWithHDFS -Pflink-runner -Dexec.args="--runner=FlinkRunner
--filesToStage=target/flinkBeam-2.2.0-SNAPSHOT-shaded.jar --flinkMaster=localhost:6123 "

```

it seems the flink cluster regard the hdfs as local file system.
The input log from flink-jobmanger.log is:

```

2017-09-27 20:17:37,962 INFO org.apache.flink.runtime.jobmanager.JobManager - Successfully
ran initialization on master in 136 ms.
2017-09-27 20:17:37,968 INFO org.apache.beam.sdk.io.FileBasedSource
- {color:red}Filepattern hdfs://localhost:9000/tmp/words2 matched 0 files with
total size 0{color}
2017-09-27 20:17:37,968 INFO org.apache.beam.sdk.io.FileBasedSource
- Splitting filepattern hdfs://localhost:9000/tmp/words2 into bundles of size 0 took 0 ms
and produced 0 files a
nd 0 bundles

```

The output error message is :

```

Caused by: java.lang.ClassCastException: {color:red}org.apache.beam.sdk.io.hdfs.HadoopResourceId cannot
be cast to org.apache.beam.sdk.io.LocalResourceId{color}
at org.apache.beam.sdk.io.LocalFileSystem.create(LocalFileSystem.java:77)

at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:256)
at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:243)

at org.apache.beam.sdk.io.FileBasedSink$Writer.open(FileBasedSink.java:922)
at org.apache.beam.sdk.io.FileBasedSink$Writer.openUnwindowed(FileBasedSink.java:884)

at org.apache.beam.sdk.io.WriteFiles.finalizeForDestinationFillEmptyShards(WriteFiles.java:909)

at org.apache.beam.sdk.io.WriteFiles.access$900(WriteFiles.java:110)
at org.apache.beam.sdk.io.WriteFiles$2.processElement(WriteFiles.java:858)

```

can somebody help me, i've try all the way just can't work it out [cry]
https://issues.apache.org/jira/browse/BEAM-2457

Imported from Jira [BEAM-2995](https://issues.apache.org/jira/browse/BEAM-2995). Original Jira may contain additional context.
Reported by: huangjianhuang.

Start with the Beam FileSystems, HadoopResourceId, LocalResourceId, and FileBasedSink stack frames, then compare behavior between the local and Flink cluster commands shown. Done means the cluster can read from and write to the hdfs:// paths without treating them as local files or raising the reported ClassCastException.

Written by the indexing model from the issue text.