Flink (1.11.2) - Can't find implementation for S3 despite correct plugins being set. Using JDK11 and Scala 2.12.11

user_1480762 0
Viewed 1313

I'm running a Flink standalone cluster with a single node using Docker in Linux. I've been running a previous version for a while in production with Flink 1.10.0 and JDK8, I was able to get S3 running properly there. Now I'm trying to update to a newer version, running Docker on my dev machine using a local S3 implementation. No matter what I try, this error keeps popping up:

org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'.

It would seem that the S3 scheme isn't being mapped to the appropriate classes. I'm positive that the right plugins are being picked up by Flink. I have the following dependencies:

val testDependencies = Seq(
  "org.scalatest" %% "scalatest" % "3.2.0" % "test"
)

val miscDependencies = Seq(
  "com.github.tototoshi" %% "scala-csv" % "1.3.6",
  "org.lz4" % "lz4-java" % "1.5.1",
  "org.json4s" %% "json4s-jackson" % "3.6.1",
  "org.apache.hadoop" % "hadoop-common" % "3.2.1",
  "redis.clients" % "jedis" % "2.9.0",
  "com.googlecode.plist" % "dd-plist" % "1.21",
  "com.couchbase.client" % "java-client" % "2.7.14",
  "org.apache.parquet" % "parquet-avro" % "1.11.1",
)

val flinkDependencies = Seq(
  "org.apache.flink" %% "flink-scala" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "provided",
  "org.apache.flink" % "flink-s3-fs-hadoop" % flinkVersion % "provided",
  "org.apache.flink" % "flink-metrics-dropwizard" % flinkVersion,
  "org.apache.flink" % "flink-formats" % flinkVersion pomOnly(),
  "org.apache.flink" % "flink-compress" % flinkVersion,
  "org.apache.flink" %% "flink-statebackend-rocksdb" % flinkVersion,
  "org.apache.flink" %% "flink-clients" % flinkVersion,
  "org.apache.flink" %% "flink-parquet" % flinkVersion
)

I confirm that I'm following the documentation to the letter.

flink-streaming apache-flink
1 Answers

After struggling with this for a while I was able to solve the problem. I'm leaving my solution here in case anyone has the same issue.

Plugin classes, such as the S3 file system factory, are detected once the jobmanager and taskmanager starts, however, they're not loaded. In my setup, the classes must be loaded dynamically once the job starts. You can find more information about how Flink loads its classes here.

As explained here, the cue to load a class is given by the existence of a file in META-INF/services inside the job's jar. For the S3 plugins to work, you need to have the file:

META-INF/services/org.apache.flink.core.fs.FileSystemFactory

which contains one line for each class that Flink should load dynamically as dependencies to your job. For example:

org.apache.flink.fs.s3hadoop.S3FileSystemFactory
org.apache.flink.fs.s3hadoop.S3AFileSystemFactory

I'm using sbt assembly to create a far JAR with my job. In my project dependencies I was including flink-s3-fs-hadoop as a provided dependency, which prevented the correct services files from being included. Once I removed that qualifier, the correct services were created and everything worked.

user_14807620
Related