apache/beam · error · UnsupportedOperationException

Not yet implemented.

Error message

Not yet implemented.

What it means

FileAwareFactoryFn supports file (MRS://) and secret placeholders in Kafka config values; a config value referencing a secret *file* hits a branch that immediately throws UnsupportedOperationException('Not yet implemented.').

Solutions

1. Use the supported inline secret placeholder syntax instead of a secret-file reference
2. Resolve the secret file yourself in a pre-processing step and inject the resolved value into the config
3. Check the Beam version for added support, or file/patch support upstream

Example fix

// before
"password": "{{secretfile://bucket/secret.txt}}" // unsupported
// after
"password": "{{secret:projects/p/secrets/db-pass}}" // supported placeholder

if (value.startsWith("{{secretfile:")) { throw new IllegalArgumentException("secret-file references are not supported by FileAwareFactoryFn"); }

boolean usesUnsupportedSecretFile(String v) { return v != null && v.contains("secretfile"); }

try { apply(configValue); } catch (UnsupportedOperationException e) { /* rewrite config to inline secret placeholder or pre-resolved value */ }

When it happens

Trigger: Any Kafka configuration value processed by this factory contains a secret-file placeholder (the secretFile branch), e.g. pointing ssl.keystore password at a secret stored as a file reference.

Common situations: Users migrating file-based secret configs (secret stored in a file referenced remotely) assume the same placeholder syntax works for secrets; it is not implemented in this Beam extension version.

Related errors

• Failed to get secret.
• UnboundedSource must enable offset-based deduplication.
• Accumulators not supported in this context
• Adding required columns is not yet supported. Encountered…
• Aggregator [ ] is not supported

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

Appendix: source

Thrown at sdks/java/extensions/kafka-factories/src/main/java/org/apache/beam/sdk/extensions/kafka/factories/FileAwareFactoryFn.java:143

                  String tmpPath = replacePathWithLocal(externalPath);
                  String localPath = downloadExternalFile(externalPath, tmpPath);
                  matcher.appendReplacement(sb, Matcher.quoteReplacement(localPath));
                  LOG.info("Downloaded {} to {}", externalPath, localPath);
                } catch (IOException io) {
                  throw new IOException("Failed to download file : " + externalPath, io);
                }
              } else if (secretValue != null) {
                try {
                  String secretId = secretValue.substring(SECRET_VALUE_PREFIX.length());
                  String processedSecret =
                      processSecret(originalValue, secretId, getSecretWithCache(secretId));

                  matcher.appendReplacement(sb, Matcher.quoteReplacement(processedSecret));
                } catch (IllegalArgumentException ia) {
                  throw new IllegalArgumentException("Failed to get secret.", ia);
                }
              } else if (secretFile != null) {
                throw new UnsupportedOperationException("Not yet implemented.");
              }
            }
            matcher.appendTail(sb);
            String processedValue = sb.toString();
            processedConfig.put(key, processedValue);
          }
        } catch (IOException ex) {
          throw new RuntimeException("Failed trying to process value for key " + key + ".", ex);
        }
      }
    } catch (IOException e) {
      throw new RuntimeException("Failed trying to process extra files.", e);
    }

    return createObject(processedConfig);
  }

  /**

View on GitHub (pinned to 12126d8942)