{"record":{"id":"527b39a266abd608","repo":"apache/pulsar","slug":"batch-configs-cannot-be-found","errorCode":null,"errorMessage":"Batch Configs cannot be found","messagePattern":"Batch Configs cannot be found","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java","lineNumber":120,"sourceCode":"        intermediateTopicConsumer.acknowledgeAsync(currentTask.getMessageId()).exceptionally(throwable -> {\n          log.error()\n                  .attr(\"messageId\", currentTask.getMessageId())\n                  .exception(throwable)\n                  .log(\"Encountered error when acknowledging completed task\");\n          setCurrentError(throwable);\n          return null;\n        });\n        currentTask = null;\n      } else {\n        return retval;\n      }\n    }\n  }\n\n  private void getBatchSourceConfigs(Map<String, Object> config) {\n    if (!config.containsKey(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY)\n      || !config.containsKey(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY)) {\n      throw new IllegalArgumentException(\"Batch Configs cannot be found\");\n    }\n\n    String batchSourceConfigJson = (String) config.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);\n    this.batchSourceConfig = new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class);\n    this.batchSourceClassName = (String) config.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);\n  }\n\n  private void initializeBatchSource() {\n    // First init the batchsource\n    ClassLoader clsLoader = Thread.currentThread().getContextClassLoader();\n    Object userClassObject = Reflections.createInstance(\n      batchSourceClassName,\n      clsLoader);\n    if (userClassObject instanceof BatchSource) {\n      @SuppressWarnings(\"unchecked\") // type parameter is erased at runtime\n      BatchSource<T> typedBatchSource = (BatchSource<T>) userClassObject;\n      batchSource = typedBatchSource;\n    } else {","sourceCodeStart":102,"sourceCodeEnd":138,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java#L102-L138","documentation":"BatchSourceExecutor.getBatchSourceConfigs validates that the source config map contains both the batch source config JSON (BATCHSOURCE_CONFIG_KEY) and the discovery triggerer class name (BATCHSOURCE_CLASSNAME_KEY). Thrown as IllegalArgumentException when either key is missing.","triggerScenarios":"Opening a batch source whose user config Map<String,Object> lacks \"__BATCHSOURCECONFIGS__\" or \"__BATCHSOURCECLASSNAME__\" keys — i.e. the function was not registered through the batch source submission flow.","commonSituations":"Submitting a BatchSource with plain pulsar-admin functions create instead of the batch source submission API/CLI; hand-writing function configs; client tooling from an older version that doesn't inject batch source metadata.","solutions":["Create the batch source via the proper submission path (pulsar-admin sources create with a batch source, or the Functions API used by the batch source tooling) so config keys are injected.","Verify the config map contains both BATCHSOURCE_CONFIG_KEY and BATCHSOURCE_CLASSNAME_KEY entries.","Upgrade pulsar-client-admin / CLI to a version matching the broker."],"exampleFix":"// before: plain function creation without batch metadata\npulsar-admin functions create --classname com.example.MyBatchSource ...\n// after: use the batch source submission tooling / include in userConfig\n--user-config \"__BATCHSOURCECONFIGS__={...},__BATCHSOURCECLASSNAME__=com.example.MyBatchSource$Builder\"","handlingStrategy":"validation","validationCode":"Map<String,Object> cfg = sourceConfig.getConfigs();\nObjects.requireNonNull(cfg.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY),\n  \"missing \" + BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);\nObjects.requireNonNull(cfg.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY),\n  \"missing \" + BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);","typeGuard":"boolean hasBatchKeys(Map<String,Object> cfg) {\n  return cfg != null\n    && cfg.containsKey(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY)\n    && cfg.containsKey(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);\n}","tryCatchPattern":"try {\n  executor.open(cfg);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().equals(\"Batch Configs cannot be found\")) {\n    // resubmit via the batch source submission API\n  }\n}","preventionTips":["Always submit batch sources through the batch source submission tooling","Never hand-edit generated function configs","Keep CLI/client versions aligned with the broker"],"tags":["batch-source","config","validation","pulsar-functions"],"backgroundTag":"missing-config-key","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}