Hi Piotr Nowojski,
I have move my filesystem plugin to FLINK_HOME/pulgins in flink 1.9.1. The jobmanage don’t start up ,and It load the filesystem plugin in my owner plugin jar . and the log is :
“2019-12-19 10:58:32,394 WARN org.apache.flink.configuration.Configuration - Config uses deprecated configuration key 'high-availability.zookeeper.storageDir' instead of proper key 'high-availability.storageDir'
2019-12-19 10:58:32,398 INFO com.filesystem.plugin.FileSystemFactoryEnhance - trying to get hadoopEnv, hadoopPath = /conf/hadoop_conf 2019-12-19 10:58:32,434 WARN org.apache.hadoop.conf.Configuration - /tmp/mammut-core-site.xml:an attempt to override final parameter: fs.defaultFS; Ignoring. 2019-12-19 10:58:32,436 WARN org.apache.hadoop.conf.Configuration - /tmp/mammut-hdfs-site.xml:an attempt to override final parameter: dfs.datanode.data.dir; Ignoring. 2019-12-19 10:58:32,436 WARN org.apache.hadoop.conf.Configuration - /tmp/mammut-hdfs-site.xml:an attempt to override final parameter: dfs.datanode.failed.volumes.tolerated; Ignoring. 2019-12-19 10:58:32,436 WARN org.apache.hadoop.conf.Configuration - /tmp/mammut-hdfs-site.xml:an attempt to override final parameter: dfs.namenode.name.dir; Ignoring. 2019-12-19 10:58:32,878 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Shutting YarnJobClusterEntrypoint down with application status FAILED. Diagnostics java.io.IOException: Could not create FileSystem for highly available storage (high-availability.storageDir) at org.apache.flink.runtime.blob.BlobUtils.createFileSystemBlobStore(BlobUtils.java:119) at org.apache.flink.runtime.blob.BlobUtils.createBlobStoreFromConfig(BlobUtils.java:92) at org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createHighAvailabilityServices(HighAvailabilityServicesUtils.java:120) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.createHaServices(ClusterEntrypoint.java:292) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.initializeServices(ClusterEntrypoint.java:257) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:202) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:164) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:163) at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:501)
at org.apache.flink.yarn.entrypoint.YarnJobClusterEntrypoint.main(YarnJobClusterEntrypoint.java:102)"
On 12/19/2019 00:01，Piotr Nowojski<pi...@ververica.com> wrote：
Hi,As Yang Wang pointed out, you should use the new plugins mechanism.If it doesn’t work, first make sure that you are shipping/distributing the plugins jars correctly - the correct plugins directory structure both on the client machine. Next make sure that the cluster has the same correct setup. This is especially true for the standalone/cluster execution modes. For yarn, mesos, docker the plugins dir should be shipped to the cluster by Flink itself, however Plugins support in yarn is currently semi broken [1]. This is already fixed, but waiting to be released in 1.9.2 and 1.10.If it still doesn’t work, look for TaskManager logs what plugins/file systems are being loaded during the startup. If none, that's the problem.Piotrek[1] https://issues.apache.org/jira/browse/FLINK-14382On 18 Dec 2019, at 12:40, Yang Wang <danrtsey...@gmail.com> wrote:You could have a try the new plugin mechanism.Create a new directory named "myhdfs" under $FLINK_HOME/plugins, and then put your filesystem related jars in it.Different plugins will be loaded by separate classloader to avoid conflict.Best,Yangvino yang <yanghua1...@gmail.com> 于2019年12月18日周三 下午6:46写道：Hi ouywl,META-INF/services/org.apache.flink.core.fs.FileSystemFactorywhich contains the class name of “ FileSystemFactoryEnhance.class”And the job mainclass is :“ public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(60*1000);env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);env.getConfig().enableSysoutLogging();Properties props = new Properties();props.put("bootstrap.servers", SERVERS);props.put("enable.auto.commit", "true");// props.put("auto.commit.interval.ms", "1000");props.put("auto.offset.reset", "latest");props.put("key.deserializer", org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StringDeserializer.class.getName());props.put("value.deserializer", StringDeserializer.class.getName());FlinkKafkaConsumer010 consumer011 = new FlinkKafkaConsumer010<String>("zyf_test_2", new SimpleStringSchema(), props);DataStream<String> source = env.addSource(consumer011).setParallelism(1);source.print();Thread.currentThread().getContextClassLoader();StreamingFileSink sink = StreamingFileSink.build();source.addSink(sink);env.execute();}”And start the job, the jobmanager filesystem is error, the log means the jobmananger use “FileSystemFactoryEnhance” filesystem and confict.As the url https://ci.apache.org/projects/flink/flink-docs-stable/ops/filesystems/#pluggable-file-systems how to avoid use “Thread.currentThread().getContextClassLoader()"
