FEATURE: Add CompletableFuture Pipeline API

[oliviarla]

🔗 Related Issue

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

  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에 값을 추가할 수 있도록 한다.

[uhm0311]

value 부분에 \\s+pipe\\r\\n가 들어갈 가능성이 있습니다.
if문에서 endsWith()으로 검사하는 로직을 추가합시다.

[uhm0311]

여기에 op 객체 말고 responseIndex를 그냥 넘기면 안되나요?

[uhm0311]

List<Object>로 반환하는 것보다 PipelineResult 클래스를 만들어서 List<Boolean> 필드와 Exception 필드를 넣어서 반환하는 것이 낫지 않나요?
결과를 받아서 사용하는 쪽에서 매번 타입 체크하고 타입 변환을 하는 것보다 나을 것 같습니다.