[BEAM-7720] Fix the exception type of InMemoryJobService when job id not found

[chadrik]

Everywhere that we call getInvocation() we should handle the possible StatusException

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[chadrik]

R: @lukecwik

[chadrik]

Original discussion about this problem is here: #8977 (comment)

[lukecwik]

Can you add the try/catch StatusRuntimeException/StatusException to all the other service calls?

[lukecwik]

Suggested change

	} catch (StatusException e) {
	} catch (StatusRuntimeException | StatusException e) {

Here and everywhere else otherwise the catch Exception will also convert the StatusRuntimeException to an INTERNAL status error.

[chadrik]

I updated StatusException to StatusRuntimeException | StatusException everywhere, but I'm a bit unclear on what's actually throwing the StatusRuntimeException. I searched the code for StatusRuntimeException and I did not see a clear path from any of those cases to here.

For example, looking at this code I didn't see anything suspicious inside the try/catch block:

  @Override
  public void getState(
      GetJobStateRequest request, StreamObserver<GetJobStateResponse> responseObserver) {
    LOG.trace("{} {}", GetJobStateRequest.class.getSimpleName(), request);
    String invocationId = request.getJobId();
    try {
      JobInvocation invocation = getInvocation(invocationId);
      JobState.Enum state = invocation.getState();
      GetJobStateResponse response = GetJobStateResponse.newBuilder().setState(state).build();
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    } catch (StatusRuntimeException | StatusException e) {
      responseObserver.onError(e);
    } catch (Exception e) {
      String errMessage =
          String.format("Encountered Unexpected Exception for Invocation %s", invocationId);
      LOG.error(errMessage, e);
      responseObserver.onError(Status.INTERNAL.withCause(e).asException());
    }
  }

[lukecwik]

Since they are runtime exceptions, nothing is required to declare that they are thrown in their method signatures and can propagate up the call stack from any arbitrary location until they hit an appropriately scoped catch statement.

Without the change to also catch the StatusRuntimeException as part of the StatusException, they would be caught as part of the Exception block and converted to INTERNAL errors.