apache/pulsar · critical · RuntimeException

error writing discovered task to intermediate topic

Error message

error writing discovered task to intermediate topic

What it means

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.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java:200

      } finally {
        discoverInProgress = false;
      }
    });
  }

  private void taskEater(String discoveredEvent, byte[] task) {
    try {
      Map<String, String> properties = new HashMap<>();
      properties.put("discoveredEvent", discoveredEvent);
      properties.put("produceTime", String.valueOf(System.currentTimeMillis()));
      TypedMessageBuilder<byte[]> message = sourceContext.newOutputMessage(intermediateTopicName, Schema.BYTES);
      message.value(task).properties(properties);
      // Note: we can only make this send async if the api returns a future to
      // the connector so that errors can be handled by the connector
      message.send();
    } catch (Exception e) {
      log.error().exception(e).log("Error writing discovered task to intermediate topic");
      throw new RuntimeException("error writing discovered task to intermediate topic");
    }
  }

  private void prepareInternal(Message<byte[]> task) {
    try {
      batchSource.prepare(task.getValue());
    } catch (Exception e) {
      log.error().exception(e).log("Error on prepare");
      throw new RuntimeException(e);
    }
  }

  @Override
  public void close() throws Exception {
    this.stop();
  }

  private void stop() throws Exception {

View on GitHub (pinned to 820761864e)

Solutions

1. Check the broker connectivity and health from the function instance; restart the function once the broker is reachable
2. Verify the intermediate topic exists or that allowAutoTopicCreation is enabled on the namespace
3. Confirm the function's auth token/role has produce permission on the intermediate topic
4. Inspect the logged underlying exception (Error writing discovered task to intermediate topic) for the root cause (e.g. TopicNotFound, ProducerBusy, MessageTooLarge)
5. Re-run the batch source connector; discovery is retried by the connector on failure

Example fix

// before (symptom)
try {
    message.send();
} catch (Exception e) {
    log.error().exception(e).log("Error writing discovered task to intermediate topic");
    throw new RuntimeException("error writing discovered task to intermediate topic");
}
// after (operator-side: ensure topic exists and broker is up)
pulsar-admin topics create persistent://public/default/__transactions_intermediate;
pulsar-admin topics grant-permissions persistent://public/default/__transactions_intermediate --role functions-role --actions produce

// before starting the batch source, ensure the intermediate topic is writable
Producer<byte[]> p = client.newProducer()
    .topic(intermediateTopic)
    .create(); // throws early if topic/permission problems exist
p.close();

try {
    message.send();
} catch (PulsarClientException e) {
    // inspect e: TopicNotFound / ProducerBusy / NotAllowed -> fix broker side, then retry
    throw new RuntimeException("intermediate topic send failed: " + e.getMessage(), e);
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Error starting LogTopic Producer for function %s
• Unable to initialize crypto config %s
• Failed to create Producer for topic ${topicName} producerNam
• Batch Configs cannot be found
• BatchSource does not implement the correct interface

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/4445adeb87fb2443. Report an issue: GitHub.