RATIS-2603. Add DataStreamApi.Resolver for group-independent read-only requests

[peterxcli]

What changes were proposed in this pull request?

RATIS-2603
Users such as Ozone may archive part of their data and mark it immutable. In Ozone, closed containers use a synthetic read pipeline whose ID does not correspond to a live Ratis group, causing DataStream reads to fail with GroupMismatchException.

Introducing a new DataStreamReadResolver, which should provide an optional path to bypass group lookup and let the application define how such reads are handled. If the resolver declines the request, the existing Ratis read path remains unchanged.

How could Users Use this?

for example, in ozone, we register our resolver to DataStreamReadResolver in XceiverServerRatis#newXceiverServerRatis when OzoneContainer starts.

public static XceiverServerRatis newXceiverServerRatis(...){
  // ...
  ClosedContainerReadResolver resolver = new ClosedContainerReadResolver(dispatcher, 
    containerController, datanodeDetails);
  DataStreamReadResolver.set(parameters, resolver);
  // ...
}

and the resolver implmentaion should be like:

/** Resolves group-independent reads against immutable local containers. */
final class ClosedContainerReadResolver implements DataStreamReadResolver {
  private final ContainerDispatcher dispatcher;
  private final ContainerController containerController;
  private final String datanodeUuid;

  ClosedContainerReadResolver(ContainerDispatcher dispatcher,
      ContainerController containerController, DatanodeDetails datanode) {
    this.dispatcher = dispatcher;
    this.containerController = containerController;
    this.datanodeUuid = datanode.getUuidString();
  }

  @Override
  public StateMachine.DataApi resolve(RaftClientRequest request)
      throws IOException {
    final ContainerCommandRequestProto requestProto = ContainerCommandRequestMessage.toProto(
        request.getMessage().getContent(), request.getRaftGroupId());
    if (requestProto.getCmdType() != Type.ReadBlock
        || requestProto.hasDatanodeUuid()
            && !datanodeUuid.equals(requestProto.getDatanodeUuid())) {
      return null;
    }

    final Container<?> container =
        containerController.getContainer(requestProto.getContainerID());
    if (container == null || container.getContainerState() != CLOSED) {
      return null;
    }

    return new StateMachine.DataApi() {
      @Override
      public void query(Message ignored, WritableByteChannel stream) {
        try {
          ContainerStateMachine.streamReadBlock(
              dispatcher, requestProto, stream);
        } catch (IOException e) {
          throw new CompletionException(e);
        }
      }
    };
  }
}

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2603

How was this patch tested?

(Please explain how this patch was tested. Ex: unit tests, manual tests)
(If this patch involves UI changes, please attach a screen-shot; otherwise, remove this)

[peterxcli]

apache/ozone#10743 is the patch that use the resolver.

cc @szetszwo @amaliujia, please could you take a look at them whenever you have time, thanks!

[amaliujia]

Nit: you can add @Nullable notation to Parameters

[amaliujia]

Why we do not choose to fall back to division when see exceptions from the resolver>

[peterxcli]

for example, the container is closed, it doesn't have the division anymore, if the resolve process thrown exception, then the fallback to division would just get the the GroupMismatchException

[szetszwo]

@peterxcli , thanks a lot for working on this. It is a really great idea!

Let's do a refactoring so that the new API is not in StateMachine; filed RATIS-2604. We could also make it similar to the FileChannel.transferTo(..) method. I have tried it a little bit; see https://issues.apache.org/jira/secure/attachment/13083178/1516_review.patch

/** Server side data stream API. */
public interface DataStreamApi { // TODO: add this RATIS-2604
  /**
   * Similar to {@link java.nio.channels.FileChannel#transferTo(long, long, WritableByteChannel)}
   * except that the parameter of this method is request message
   * but not position, count in FileChannel.
   *
   * @param request the request message
   * @param stream the output stream to send the results
   * @return the number of bytes transferred
   */
  long transferTo(Message request, WritableByteChannel stream) throws IOException;

  /** For resolving {@link DataStreamApi}. */
  @FunctionalInterface
  interface Resolver { // TODO: add this in RATIS-2603
    /** @return the data API handling this request, or null if the API is not found. */
    DataStreamApi resolve(RaftClientRequest request);
  }
}

[szetszwo]

+1 the change looks good.

[peterxcli]

Thanks @szetszwo for the review, refactoring suggestion and renaming title! @amaliujia for the review!