FEATURE: Add CompletableFuture Pipeline API

[oliviarla]

🔗 Related Issue

• Pipeline API를 추가합니다.
• 사용 방법
  • Pipeline 객체를 생성하여 하나의 파이프라인으로 포함할 명령들을 조합한다.
  • 파이프라인에 포함할 수 있는 명령들을 제한할 수 있다. (명령어의 종류, 개수)
  • 명령어의 개수는 최대 500개로 제한한다. 이를 통해 한 노드에는 최대 하나의 파이프라인이 보내진다.
  • AsyncArcusCommands 객체 전역적으로 flush 여부를 설정하지 않으므로 세밀하게 파이프라인에 포함할 명령을 지정할 수 있다.

  public class Pipeline<V> {
  
      // 다양한 명령을 포함할 수 있는 PipelineOperationImpl 타입을 value에 저장한다.
      private Map<MemcachedNode, Operation> ops;
  
      public Pipeline() {
        // ...
      }
  
      public Pipeline<V> lopInsert(String key, int index, V value) {
        // ...
        return this;
      }
  
      public Pipeline<V> mopInsert(String key, String mkey, V value) {
        // ...
        return this;
      }
  
      // ...
  
  }
  

  • 명령의 조합 순서대로 파이프라인이 만들어지지만, 서버에서 처리 도중 다른 연산이 끼어들 수 있다.
  • 서로 다른 여러 키를 파이프라인에 포함하는 경우, 각 노드 별로 파이프라인이 생성되어 노드에 전달된다.
  • Pipeline 객체를 execute 메서드에 넘겨야 파이프라인이 수행되며, List<Object> 에 각 명령에 대한 응답이 담긴다.

  ArcusFuture<List<Object>> execute(Pipeline<T> pipeline)
  

⌨️ What I did

• 예외 처리
  • 모든 응답이 올 때 까지 기다린 후 여러 요청에서 발생한 실패/에러들을 하나의 CompositeException 타입으로 모아 던진다.
• switchover 처리
  • responseIndex를 사용하여 아직 응답이 오지 않은 operation들만 새로운 master 노드에서 initialize할 때 포함되도록 한다.
• migration 처리
  • gotStatus 메서드의 인자로 Operation을 받아, 여러 Operation으로 분리되더라도 원본 콜백에서 원본 index에 값을 추가할 수 있도록 한다.