Remote cluster submission from Flink Runner broken due to staging issues

[kennknowles]

The usual way to submit pipelines with the Flink Runner is to build a fat jar and use the bin/flink utility to submit the jar to a Flink cluster. This works fine.

Alternatively, the Flink Runner can use the flinkMaster pipeline option to specify a remote cluster. Upon submitting an example we get the following at Flink's JobManager.

Caused by: java.lang.IllegalAccessError: class sun.reflect.GeneratedSerializationConstructorAccessor70
cannot access its superclass sun.reflect.SerializationConstructorAccessorImpl
	at sun.misc.Unsafe.defineClass(Native
Method)
	at sun.reflect.ClassDefiner.defineClass(ClassDefiner.java:63)
	at sun.reflect.MethodAccessorGenerator$1.run(MethodAccessorGenerator.java:399)
	at
sun.reflect.MethodAccessorGenerator$1.run(MethodAccessorGenerator.java:394)
	at java.security.AccessController.doPrivileged(Native
Method)
	at sun.reflect.MethodAccessorGenerator.generate(MethodAccessorGenerator.java:393)
	at sun.reflect.MethodAccessorGenerator.generateSerializationConstructor(MethodAccessorGenerator.java:112)
	at
sun.reflect.ReflectionFactory.newConstructorForSerialization(ReflectionFactory.java:340)
	at java.io.ObjectStreamClass.getSerializableConstructor(ObjectStreamClass.java:1420)
	at
java.io.ObjectStreamClass.access$1500(ObjectStreamClass.java:72)
	at java.io.ObjectStreamClass$2.run(ObjectStreamClass.java:497)
	at
java.io.ObjectStreamClass$2.run(ObjectStreamClass.java:472)
	at java.security.AccessController.doPrivileged(Native
Method)
	at java.io.ObjectStreamClass.<init>(ObjectStreamClass.java:472)
	at java.io.ObjectStreamClass.lookup(ObjectStreamClass.java:369)
	at
java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:598)
	at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1630)
	at
java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1521)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1781)
	at
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
	at
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
	at
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
	at java.io.ObjectInputStream.readObject(ObjectInputStream.java:373)
	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:502)
	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:489)
	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:477)
	at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:438)
	at
org.apache.flink.runtime.operators.util.TaskConfig.getStubWrapper(TaskConfig.java:288)
	at org.apache.flink.runtime.jobgraph.InputFormatVertex.initializeOnMaster(InputFormatVertex.java:63)
	...
32 more

It appears there is an issue with the staging via PipelineResources.

Imported from Jira BEAM-7478. Original Jira may contain additional context.
Reported by: mxm.