{"record":{"id":"431545985896f6bf","repo":"apache/beam","slug":"expected-state-to-be-started-but-was-complete-error","errorCode":null,"errorMessage":"Expected state to be STARTED, but was COMPLETE_ERROR","messagePattern":"Expected state to be STARTED, but was COMPLETE_ERROR","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java","lineNumber":206,"sourceCode":"        case COMPLETE_SUCCESS:\n          throw new IllegalStateException(\n              \"Expected state to be PENDING or STARTED, but was COMPLETE_SUCCESS\");\n        case COMPLETE_ERROR:\n          throw new IllegalStateException(\n              \"Expected state to be PENDING or STARTED, but was COMPLETE_ERROR\");\n      }\n    }\n\n    public void checkStarted() {\n      switch (this) {\n        case STARTED:\n          return;\n        case PENDING:\n          throw new IllegalStateException(\"Expected state to be STARTED, but was PENDING\");\n        case COMPLETE_SUCCESS:\n          throw new IllegalStateException(\"Expected state to be STARTED, but was COMPLETE_SUCCESS\");\n        case COMPLETE_ERROR:\n          throw new IllegalStateException(\"Expected state to be STARTED, but was COMPLETE_ERROR\");\n      }\n    }\n  }\n\n  private abstract class BaseRpcAttempt implements RpcAttempt {\n    private final Logger logger;\n    final O11y o11y;\n    final StatusCodeAwareBackoff backoff;\n    final Sleeper sleeper;\n\n    AttemptState state;\n    Instant start;\n\n    @SuppressWarnings(\n        \"initialization.fields.uninitialized\") // allow transient fields to be managed by component\n    // lifecycle\n    BaseRpcAttempt(Context context, O11y o11y, StatusCodeAwareBackoff backoff, Sleeper sleeper) {\n      this.logger = LoggerFactory.getLogger(String.format(\"%s.RpcQos\", context.getNamespace()));","sourceCodeStart":188,"sourceCodeEnd":224,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java#L188-L224","documentation":"RpcQosImpl tracks the lifecycle of a Firestore RPC attempt through an internal state machine (STARTED, PENDING, COMPLETE_SUCCESS, COMPLETE_ERROR). Attempt-scoped methods like getMsToNextRttBreaker check via checkStarted() that the attempt is still in the STARTED state before acting. This IllegalStateException is thrown when the method is called on an attempt that already completed with an error, meaning the caller is reusing a finished attempt instead of starting a new one.","triggerScenarios":"Calling a method on an RpcAttempt (from RpcQos.newAttempt) after a prior call on the same attempt threw an exception and moved the attempt into the COMPLETE_ERROR state — typically retry logic that keeps using the old attempt object instead of requesting a fresh one.","commonSituations":"Custom retry loops around Firestore RPCs that catch a GoogleApiException but retry on the same attempt; writing a custom DoFn that caches a single RpcAttempt across elements; upgrading Beam versions where the QoS layer began enforcing attempt state transitions strictly.","solutions":["Obtain a new attempt via rpcQos.newAttempt() inside the retry loop for every retry, instead of reusing the failed attempt","Inspect the stack trace to find which method was called after the attempt was already aborted/failed and move that call before the failure point","Ensure awaitOutOfBandPermission/next attempt logic follows the pattern in Beam's Firestore V1 connector (see RpcQosImpl usage in FirestoreV1Fn)","Update to a newer Beam version in case the error stems from a fixed QoS state-handling bug"],"exampleFix":"// before\nRpcAttempt attempt = qos.newAttempt();\nfor (int i = 0; i < retries; i++) {\n  try { attempt.pause(); doRpc(); break; } catch (Exception e) { /* reuse attempt */ }\n}\n// after\nfor (int i = 0; i < retries; i++) {\n  RpcAttempt attempt = qos.newAttempt();\n  try { attempt.checkStarted(); doRpc(); break; } catch (Exception e) { /* fresh attempt next loop */ }\n}","handlingStrategy":"try-catch","validationCode":"// before using an attempt, guard on state\nif (attempt != null && isStarted(attempt)) { attempt.pause(); ... }\n// implement isStarted by tracking state in your wrapper since state is private","typeGuard":"boolean isFreshAttempt(RpcAttempt a) { return a instanceof RpcQosImpl.RpcWriteAttempt || /* track via wrapper flag */ attemptUsed == false; }","tryCatchPattern":"try {\n  attempt.pause();\n  // rpc work\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"Expected state to be STARTED\")) {\n    attempt = rpcQos.newAttempt(); // get a fresh attempt and retry\n  } else throw e;\n}","preventionTips":["Create a new RpcAttempt per try iteration — never reuse an attempt after a failure","Wrap attempt usage in a small helper class that marks the attempt consumed after any exception","Follow the exact pattern in Beam's FirestoreV1Fn implementation"],"tags":["java","firestore","state-machine","rpc"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}