apache/beam · error · RuntimeException

Unable to construct FactoryFn

Error message

Unable to construct FactoryFn %s: %s

What it means

setupExternalBuilder catches any exception while instantiating the configured consumer FactoryFn (e.g. KerberosConsumerFactoryFn via InstanceBuilder) and rethrows it as a RuntimeException with the class name and original message. It signals that the factory function class could not be constructed with the given parameters.

Solutions

1. Read the wrapped cause ('...: <message>') to find the root failure.
2. Verify consumerFactoryFnClass is the correct fully-qualified class name and is included in the job's classpath/staged files.
3. Check consumerFactoryFnParams types (krb5Location must be a String) and values (valid GCS/Secret Manager path).
4. Ensure credentials/permissions allow reading the krb5.conf location.

Example fix

// before
{"consumerFactoryFnClass": "com.example.MyFactory"} // not staged
// after
{"consumerFactoryFnClass": "org.apache.beam.sdk.io.kafka.KerberosConsumerFactoryFn", "consumerFactoryFnParams": {"krb5Location": "gs://bucket/krb5.conf"}}

Class.forName(config.consumerFactoryFnClass);

try { build(cfg); } catch (RuntimeException e) { log(e.getCause()); }

When it happens

Trigger: The consumerFactoryFnClass cannot be loaded or instantiated: class not on classpath, constructor signature mismatch with the provided params, invalid krb5Location (download failure inside the factory), or any exception thrown by the factory's constructor.

Common situations: Typo in the fully-qualified class name; custom factory not staged with the job; wrong parameter types in consumerFactoryFnParams (e.g. non-String krb5Location); unreachable GCS/Secret Manager krb5.conf.

Related errors

• consumerPollingTimeout should be > 0.
• timestampPolicy should be one of (ProcessingTime…
• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• Could not find class:

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/978639ff3f9f250e. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:977

        if (config.consumerFactoryFnClass != null) {
          if (config.consumerFactoryFnClass.contains("KerberosConsumerFactoryFn")) {
            try {
              if (!config.consumerFactoryFnParams.containsKey("krb5Location")) {
                throw new IllegalArgumentException(
                    "The KerberosConsumerFactoryFn requires a location for the krb5.conf file. "
                        + "Please provide either a GCS location or Google Secret Manager location for this file.");
              }
              String krb5Location = config.consumerFactoryFnParams.get("krb5Location");
              builder.setConsumerFactoryFn(
                  InstanceBuilder.ofType(
                          new TypeDescriptor<
                              SerializableFunction<
                                  Map<String, Object>, Consumer<byte[], byte[]>>>() {})
                      .fromClassName(config.consumerFactoryFnClass)
                      .withArg(String.class, Objects.requireNonNull(krb5Location))
                      .build());
            } catch (Exception e) {
              throw new RuntimeException(
                  "Unable to construct FactoryFn "
                      + config.consumerFactoryFnClass
                      + ": "
                      + e.getMessage(),
                  e);
            }
          }
        }
      }

      private static <T> Coder<T> resolveCoder(Class<Deserializer<T>> deserializer) {
        for (Method method : deserializer.getDeclaredMethods()) {
          if (method.getName().equals("deserialize")) {
            Class<?> returnType = method.getReturnType();
            if (returnType.equals(Object.class)) {
              continue;
            }
            if (returnType.equals(byte[].class)) {

View on GitHub (pinned to 12126d8942)