{"record":{"id":"471f9b9934041182","repo":"apache/beam","slug":"simulating-failure-for-outgoingmessage","errorCode":null,"errorMessage":"Simulating failure for ${outgoingMessage}","messagePattern":"Simulating failure for (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java","lineNumber":497,"sourceCode":"            expectedTopic);\n      }\n      Set<OutgoingMessage> remainingExpected =\n          checkStateNotNull(STATE.remainingExpectedOutgoingMessages);\n      Set<OutgoingMessage> remainingFailing =\n          checkStateNotNull(STATE.remainingFailingOutgoingMessages);\n      @MonotonicNonNull String batchOrderingKey = null;\n      for (OutgoingMessage outgoingMessage : outgoingMessages) {\n        if (batchOrderingKey == null) {\n          batchOrderingKey = outgoingMessage.getMessage().getOrderingKey();\n        }\n        checkState(outgoingMessage.getMessage().getOrderingKey().equals(batchOrderingKey));\n        if (isDynamic) {\n          checkState(topic.getPath().equals(outgoingMessage.topic()));\n        } else {\n          checkState(outgoingMessage.topic() == null);\n        }\n        if (remainingFailing.remove(outgoingMessage)) {\n          throw new RuntimeException(\"Simulating failure for \" + outgoingMessage);\n        }\n        checkState(\n            remainingExpected.remove(outgoingMessage),\n            \"Unexpected outgoing message %s\",\n            outgoingMessage);\n      }\n      return outgoingMessages.size();\n    }\n  }\n\n  @Override\n  public List<IncomingMessage> pull(\n      long requestTimeMsSinceEpoch,\n      SubscriptionPath subscription,\n      int batchSize,\n      boolean returnImmediately)\n      throws IOException {\n    synchronized (STATE) {","sourceCodeStart":479,"sourceCodeEnd":515,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java#L479-L515","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\nPubsubTestClient.publishFailures(topic, incomingOutgoingMessage);\n// after\nPubsubTestClient.publishMessages(topic, incomingOutgoingMessage);","handlingStrategy":"try-catch","validationCode":"// remove the message from the registered failing set before publishing\nSet<PubsubClient.OutgoingMessage> failing = PubsubTestClient.publishFailuresSet(); // test helper\nboolean willFail = failing.contains(outgoingMessage);","typeGuard":null,"tryCatchPattern":"try {\n  client.publish(topic, data, timestamp);\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().startsWith(\"Simulating failure for \")) {\n    // expected injected failure; retry or assert\n  } else {\n    throw e;\n  }\n}","preventionTips":["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"],"tags":["java","pubsub","test-failure-injection"],"backgroundTag":"simulated-publish-failure","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}