What causes "Simulating failure for ${outgoingMessage}" in apache/beam?

Remove the message from the publishFailures(...) registration if the failure is not intended

apache/beam · warning · RuntimeException

Simulating failure for

Error message

Simulating failure for ${outgoingMessage}

What it means

PubsubTestClient is a PubsubClient implementation for unit tests. publish() throws this RuntimeException when the outgoing message matches one of the pre-registered 'failing' messages configured via PubsubTestClient.publishFailures(...). It exists purely so tests can simulate a publish failure and verify retry/error handling in calling code.

Solutions

Remove the message from the publishFailures(...) registration if the failure is not intended
Change the message payload/topic so it no longer equals a registered failing message
Handle the RuntimeException in code under test if failure-injection is the point of the test
Wrap publish in try-catch in the test harness and assert on the exception

Example fix

// before
PubsubTestClient.publishFailures(topic, incomingOutgoingMessage);
// after
PubsubTestClient.publishMessages(topic, incomingOutgoingMessage);

Defensive patterns

Strategy: try-catch

Validate before calling

// remove the message from the registered failing set before publishing
Set<PubsubClient.OutgoingMessage> failing = PubsubTestClient.publishFailuresSet(); // test helper
boolean willFail = failing.contains(outgoingMessage);

Try / catch

try {
  client.publish(topic, data, timestamp);
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().startsWith("Simulating failure for ")) {
    // expected injected failure; retry or assert
  } else {
    throw e;
  }
}

Prevention

Only register messages as failing when the test explicitly verifies failure handling
Keep failing-message registrations adjacent to the publish call in the test
Clear publishFailures state between tests to avoid stale registrations

When it happens

Trigger: The test was constructed with publishFailures(topics, messages...) and code under test calls publish(topic, data, timestamp) with a message exactly equal (same topic, data, timestamp) to one registered as failing. Also thrown when the topic is dynamic and the message topic does not match, or a static-topic message has a non-null topic, via checkState.

Common situations: A Beam test that intentionally injects publish failures; a developer confused why publish suddenly fails after adding a message to the failing set; a stale failing-message registration left in the test setup.

Related errors

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java:497

            expectedTopic);
      }
      Set<OutgoingMessage> remainingExpected =
          checkStateNotNull(STATE.remainingExpectedOutgoingMessages);
      Set<OutgoingMessage> remainingFailing =
          checkStateNotNull(STATE.remainingFailingOutgoingMessages);
      @MonotonicNonNull String batchOrderingKey = null;
      for (OutgoingMessage outgoingMessage : outgoingMessages) {
        if (batchOrderingKey == null) {
          batchOrderingKey = outgoingMessage.getMessage().getOrderingKey();
        }
        checkState(outgoingMessage.getMessage().getOrderingKey().equals(batchOrderingKey));
        if (isDynamic) {
          checkState(topic.getPath().equals(outgoingMessage.topic()));
        } else {
          checkState(outgoingMessage.topic() == null);
        }
        if (remainingFailing.remove(outgoingMessage)) {
          throw new RuntimeException("Simulating failure for " + outgoingMessage);
        }
        checkState(
            remainingExpected.remove(outgoingMessage),
            "Unexpected outgoing message %s",
            outgoingMessage);
      }
      return outgoingMessages.size();
    }
  }

  @Override
  public List<IncomingMessage> pull(
      long requestTimeMsSinceEpoch,
      SubscriptionPath subscription,
      int batchSize,
      boolean returnImmediately)
      throws IOException {
    synchronized (STATE) {

View on GitHub (pinned to 12126d8942)