{"record":{"id":"4445adeb87fb2443","repo":"apache/pulsar","slug":"error-writing-discovered-task-to-intermediate-topi","errorCode":null,"errorMessage":"error writing discovered task to intermediate topic","messagePattern":"error writing discovered task to intermediate topic","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java","lineNumber":200,"sourceCode":"      } finally {\n        discoverInProgress = false;\n      }\n    });\n  }\n\n  private void taskEater(String discoveredEvent, byte[] task) {\n    try {\n      Map<String, String> properties = new HashMap<>();\n      properties.put(\"discoveredEvent\", discoveredEvent);\n      properties.put(\"produceTime\", String.valueOf(System.currentTimeMillis()));\n      TypedMessageBuilder<byte[]> message = sourceContext.newOutputMessage(intermediateTopicName, Schema.BYTES);\n      message.value(task).properties(properties);\n      // Note: we can only make this send async if the api returns a future to\n      // the connector so that errors can be handled by the connector\n      message.send();\n    } catch (Exception e) {\n      log.error().exception(e).log(\"Error writing discovered task to intermediate topic\");\n      throw new RuntimeException(\"error writing discovered task to intermediate topic\");\n    }\n  }\n\n  private void prepareInternal(Message<byte[]> task) {\n    try {\n      batchSource.prepare(task.getValue());\n    } catch (Exception e) {\n      log.error().exception(e).log(\"Error on prepare\");\n      throw new RuntimeException(e);\n    }\n  }\n\n  @Override\n  public void close() throws Exception {\n    this.stop();\n  }\n\n  private void stop() throws Exception {","sourceCodeStart":182,"sourceCodeEnd":218,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java#L182-L218","documentation":"BatchSourceExecutor's taskEater writes each discovered task (record) to the intermediate topic via the Pulsar producer's synchronous message.send(). If the send throws any Exception, the executor logs it and rethrows a RuntimeException so the connector's discovery loop fails fast. This error means the discovered task could not be persisted to the internal intermediate topic, typically because the topic/producer is unavailable or the broker rejected the message.","triggerScenarios":"Producer.send() to the intermediate topic fails during triggerDiscover: topic does not exist / auto-creation disabled, broker unreachable, producer closed, message too large, namespace/quota issues, or authentication/authorization failure on the intermediate topic.","commonSituations":"Batch source connectors (e.g. CDC, file, tiered ingestion) running while the broker restarts or the intermediate topic was deleted; broken auth config so the function can't produce to the internal topic; intermediate topic names containing characters rejected by the broker.","solutions":["Check the broker connectivity and health from the function instance; restart the function once the broker is reachable","Verify the intermediate topic exists or that allowAutoTopicCreation is enabled on the namespace","Confirm the function's auth token/role has produce permission on the intermediate topic","Inspect the logged underlying exception (Error writing discovered task to intermediate topic) for the root cause (e.g. TopicNotFound, ProducerBusy, MessageTooLarge)","Re-run the batch source connector; discovery is retried by the connector on failure"],"exampleFix":"// before (symptom)\ntry {\n    message.send();\n} catch (Exception e) {\n    log.error().exception(e).log(\"Error writing discovered task to intermediate topic\");\n    throw new RuntimeException(\"error writing discovered task to intermediate topic\");\n}\n// after (operator-side: ensure topic exists and broker is up)\npulsar-admin topics create persistent://public/default/__transactions_intermediate;\npulsar-admin topics grant-permissions persistent://public/default/__transactions_intermediate --role functions-role --actions produce","handlingStrategy":"validation","validationCode":"// before starting the batch source, ensure the intermediate topic is writable\nProducer<byte[]> p = client.newProducer()\n    .topic(intermediateTopic)\n    .create(); // throws early if topic/permission problems exist\np.close();","typeGuard":null,"tryCatchPattern":"try {\n    message.send();\n} catch (PulsarClientException e) {\n    // inspect e: TopicNotFound / ProducerBusy / NotAllowed -> fix broker side, then retry\n    throw new RuntimeException(\"intermediate topic send failed: \" + e.getMessage(), e);\n}","preventionTips":["Pre-create the intermediate topic or enable auto topic creation on the namespace","Grant the function role produce permissions on the intermediate topic","Monitor broker health/alerts so discovery doesn't run against a down broker","Check function logs for the chained exception to identify producer failures quickly"],"tags":["pulsar-functions","batch-source","producer","intermediate-topic"],"backgroundTag":"message-produce-failed","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"}