[BEAM-4291] Propagates artifact retrieval token in Flink runner and to the Java harness

[jkff]

Also changes Go artifact-related testing code to use the staging and retrieval tokens.

The bulk of this change is the changes in generated Go proto code.

R: @herohde for Go code, @angoenka for Java code

[jkff]

OK, this is fully ready for review I think.

[herohde]

High-level comment: What will we be using this token for, concretely? We already have the client ID to tell SDK harnesses apart. I don't recall it being discussed in the design doc.

It also seems somewhat out of place to include a retrieval_token here, but not the ApiServiceDescriptor.

[jkff]

This design was discussed some time ago (see also #5582) - we're coming from the assumption that all services are (or at least may be) globally distributed and stateless, i.e. we're not relying on the assumption that there's 1 ArtifactRetrievalService per worker or per harness. Without that assumption, we need the ArtifactRetrievalService calls to be somehow linked to which job we're talking about. Likewise, ArtifactStagingService also needs to know which job we're talking about.

We decided to do this by propagating tokens:

• PrepareJob returns a token used for ArtifactStagingService calls
• ArtifactStagingService.CommitManifest returns a token used for ArtifactRetrievalService calls

This token is an opaque string containing the information necessary for the service to do its job. In practice, with the "distributed file system" based implementations of both services, we're using (basically) a base path as the token.

Alternatively we could explicitly include the job ID in the RPCs, but that would require the services to do some sort of global lookup of artifact placement parameters based on job ID, it seems easier to include the necessary parameters explicitly in the token.

Now, since a retrieval token is needed for the harness to talk to the retrieval service, it seems reasonable to include it in provision info. It's not part of the service descriptor because it does not identify the service, it only gives a necessary argument for its RPCs.

[herohde]

The original design has the same assumptions. The provisioning and control service will still have to differentiate harnesses based on the client id (using internal lookups and whatnot). Artifact service is now different.

For the artifact ApiServiceDescriptor, I meant that if we must call provisioning first for the token then it might as well vendor the service descriptor, too (and perhaps the control service as well towards a smaller container contract). That would be a larger change, though.

[jkff]

Ack, we haven't thought about applying similar changes to provision and control services but it makes sense too - might do it later if we run into issues stemming from that.

[herohde]

Nit: this token should not have to be identical to the commit side.

[jkff]

(addressed by previous comment: this token is, by definition, a way to pass information from the Commit call to the retrieval service - so it has to be identical to the commit side)

[herohde]

It's not a requirement that the same underlying storage and necessary metadata is used for staging and retrieval (which might use a caching intermediate, say), so this seems like an unnecessary restriction. The user/sdk-side code may run in completely different environments.

[jkff]

I think a caching intermediate would not need its own token format - it could just pass the given token to the "real" retrieval service?..

[herohde]

Added some comments, but I don't want to block this PR if we're already going down this way.

[jkff]

Thanks, merged, we can discuss the other issues in parallel if needed.