apache/flink · error · InvalidProgramException

The implementation of the {functionType} is not serializable

Error message

The implementation of the {functionType} is not serializable. The implementation accesses fields of its enclosing class, which is a common reason for non-serializability. A common solution is to make the function a proper (non-inner) class, or a static inner class.

What it means

Thrown by ClosureCleaner after it cleaned the function and still could not serialize it, AND the bytecode analysis detected that the implementation accesses fields of its enclosing instance (the synthetic this$0 reference). A non-static inner/anonymous/local class captures its enclosing instance, and if that instance or anything reachable from it is non-serializable, the function cannot ship to the cluster. The message specifically tells you the cause is the enclosing-class capture.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/java/ClosureCleaner.java:170

                String msg =
                        functionType == null
                                ? (func + " is not serializable.")
                                : ("The implementation of the "
                                        + functionType
                                        + " is not serializable.");

                if (closureAccessed) {
                    msg +=
                            " The implementation accesses fields of its enclosing class, which is "
                                    + "a common reason for non-serializability. "
                                    + "A common solution is to make the function a proper (non-inner) class, or "
                                    + "a static inner class.";
                } else {
                    msg += " The object probably contains or references non serializable fields.";
                }

                throw new InvalidProgramException(msg, e);
            }
        }
    }

    private static boolean needsRecursion(Field f, Object fo) {
        return (fo != null
                && !Modifier.isStatic(f.getModifiers())
                && !Modifier.isTransient(f.getModifiers())
                && !canBeSerialized(fo));
    }

    private static boolean canBeSerialized(Object o) {
        try {
            InstantiationUtil.serializeObject(o);
            return true;
        } catch (Exception e) {
            return false;
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Make the function a static nested class or a top-level class so it does not capture an enclosing instance.
2. Extract the needed values into local variables captured by a lambda (effectively final) rather than reading outer instance fields.
3. Mark non-serializable fields as transient if they truly must live on the function, and reinitialize them in open().
4. Use a proper named class implementing the Flink function interface instead of an anonymous inner class.

Example fix

// before
public class Job {
  private final DbConn conn; // non-serializable
  public void run(DataStream<X> s) {
    s.map(new MapFunction<X,Y>() { // anonymous inner, captures this.conn
      public Y map(X x){ return conn.query(x); }
    });
  }
}

// after
public class Job {
  public void run(DataStream<X> s) {
    s.map(new MyMapper()); // static nested / top-level class
  }
  public static class MyMapper extends RichMapFunction<X,Y> {
    private transient DbConn conn;
    public void open(Configuration c){ conn = new DbConn(); }
    public Y map(X x){ return conn.query(x); }
  }
}

// Validate serializability before submitting
try {
    org.apache.flink.util.InstantiationUtil.serializeObject(myFunction);
} catch (Exception e) {
    throw new RuntimeException("Function is not serializable; make it static/top-level", e);
}

// Ensure the function is declared as a static nested or top-level class
// (compile-time discipline: no anonymous inner classes in non-static scope)
static boolean isStaticOrTopLevel(Class<?> c) {
    return c.getEnclosingClass() == null
        || java.lang.reflect.Modifier.isStatic(c.getModifiers());
}

try {
    org.apache.flink.api.java.ClosureCleaner.clean(func, org.apache.flink.api.common.ExecutionConfig.ClosureCleanerLevel.RECURSIVE);
} catch (org.apache.flink.api.common.InvalidProgramException e) {
    // extract message, suggest static class refactor
    throw e;
}

When it happens

Trigger: Passing an anonymous or non-static inner class that implements a Flink function (MapFunction, FilterFunction, etc.) to an operator when that class references an outer field; ClosureCleaner.clean() runs at program construction, detects this$0 access via This0AccessFinder, fails final serialization, and throws InvalidProgramException.

Common situations: Writing lambdas/method-local anonymous classes inside a non-static context (e.g. inside a non-static method of your JobMain) that capture instance fields; using a Spark-style closure in a class holding a non-serializable resource (DB connection, logger with appender); refactoring a static class into an inner class.

Related errors

• Object {obj} is not serializable
• Cannot deserialize and unwrap accumulators properly.
• Failed to serialize ExecutionPlan.
• Failed to deserialize coordination response
• Cannot deserialize and unwrap accumulators properly.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/ce79686e745822e2. Report an issue: GitHub.