Incompatible class versions when submitting Flink a job

Viewed 593

I'm trying to submit a streaming job using Beam 2.27/Flink 1.12, using the following Maven command-line:

mvn exec:java -Dexec.mainClass=org.example.MyPipelineClass -Pflink-runner -Dexec.args="--runner=FlinkRunner --flinkMaster=flink-host:8081 --filesToStage=target/pipelines-bundled-1.0.0.jar"

Things were working smoothly for a bit, but I recently made some edits and wanted to run a new version of the pipeline and am now getting the following error:

Caused by: java.io.InvalidClassException: org.apache.flink.streaming.api.graph.StreamConfig$NetworkInputConfig; local class incompatible: stream classdesc serialVersionUID = -3137689219135046939, local class serialVersionUID = 3698633776553163849
    at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:699)
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2003)
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1850)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2160)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.readArray(ObjectInputStream.java:2093)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1655)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:600)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:587)
    at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:541)
    at org.apache.flink.streaming.api.graph.StreamConfig.getInputs(StreamConfig.java:263)
    ... 21 more

My pom.xml looks something like this:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>org.example</groupId>
  <artifactId>pipelines</artifactId>
  <version>1.0.0</version>

  <packaging>jar</packaging>

  <properties>
      <beam.version>2.27.0</beam.version>
      <flink.version>1.12</flink.version>
      <maven-compiler-plugin.version>3.7.0</maven-compiler-plugin.version>
      <maven-exec-plugin.version>1.6.0</maven-exec-plugin.version>
      <maven-jar-plugin.version>3.2.0</maven-jar-plugin.version>
      <maven-shade-plugin.version>3.1.0</maven-shade-plugin.version>
  </properties>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>${maven-compiler-plugin.version}</version>
        <configuration>
          <source>1.8</source>
          <target>1.8</target>
        </configuration>
      </plugin>

      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-jar-plugin</artifactId>
        <version>${maven-jar-plugin.version}</version>
      </plugin>

      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>${maven-shade-plugin.version}</version>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <finalName>${project.artifactId}-bundled-${project.version}</finalName>
              <filters>
                <filter>
                  <artifact>*:*</artifact>
                  <excludes>
                    <exclude>META-INF/LICENSE</exclude>
                    <exclude>META-INF/*.SF</exclude>
                    <exclude>META-INF/*.DSA</exclude>
                    <exclude>META-INF/*.RSA</exclude>
                  </excludes>
                </filter>
              </filters>
              <transformers>
                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
              </transformers>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>

    <pluginManagement>
      <plugins>
        <plugin>
          <groupId>org.codehaus.mojo</groupId>
          <artifactId>exec-maven-plugin</artifactId>
          <version>${maven-exec-plugin.version}</version>
          <configuration>
            <cleanupDaemonThreads>false</cleanupDaemonThreads>
          </configuration>
        </plugin>
      </plugins>
    </pluginManagement>
  </build>

  <profiles>
    <profile>
      <id>flink-runner</id>
      <dependencies>
        <dependency>
          <groupId>org.apache.beam</groupId>
          <artifactId>beam-runners-flink-${flink.version}</artifactId>
          <version>${beam.version}</version>
          <scope>runtime</scope>
        </dependency>
      </dependencies>
    </profile>
  </profiles>

  <dependencies>
    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-core</artifactId>
      <version>${beam.version}</version>
    </dependency>

    <!-- note that the pipeline runs on GCP -->
    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
      <version>${beam.version}</version>
      <exclusions>
        <exclusion>
          <groupId>com.google.cloud.bigtable</groupId>
          <artifactId>bigtable-client-core</artifactId>
        </exclusion>
      </exclusions>
    </dependency>

    <!-- main output is JDBC/Postgres -->
    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-io-jdbc</artifactId>
      <version>${beam.version}</version>
    </dependency>
    <dependency>
      <groupId>org.postgresql</groupId>
      <artifactId>postgresql</artifactId>
      <version>42.2.4</version>
    </dependency>
  </dependencies>
</project>

Of course, before submitting the job I built the JAR using mvn clean package -Pflink-runner.

Flink is self-hosted on a GKE instance, and is deployed using the flink:1.12.4-java8 image.

I found this issue which seems to have the same error message, but no clear offender.

Any help or ideas to explore would be greatly appreciated, I have no idea what to investigate.

1 Answers

I found the issue own my own. In the pom.xml above, I specified that the Beam runner was beam-runners-flink-1.12, and the cluster was running Flink 1.12.4

Beam's Flink runner does not specify a patch, and uses Flink 1.12.0

Downgrading the cluster fixed my issue.

Related