INTERNAL: read while PIPE_ERROR received in the pipe operation

[oliviarla]

🔗 Related Issue

• https://github.com/jam2in/arcus-works/issues/616

⌨️ What I did

• 본 PR에서는 파이프 연산 요청에서 에러(CLIENT_ERROR, ERROR, SERVER_ERROR) 응답 시 PIPE_ERROR를 끝까지 읽도록 변경한다.
• 기존에는 에러 응답 시 즉시 예외를 생성해 던졌기 때문에 PIPE_ERROR를 읽지 않았다.
• 동기 파이프 연산 처리 시 에러로 인해 수행되지 않은 연산을 failedResult를 통해 알려야 하는데, 기존 형태에서는 에러 시 바로 예외를 던지므로, 에러 발생으로 인해 수행되지 않은 연산에 대해 callback.gotStatus()를 호출할 수 없다.
• 따라서 PIPE_ERROR를 읽도록 변경만 해두고, 수행되지 않은 연산에 대한 gotStatus() 호출은 INTERNAL: make lop piped operations process synchronously #795 에 추가하도록 한다.

[jhpark816]

@uhm0311 먼저 리뷰하면 됩니다.

[uhm0311]

1차 질문입니다.

[uhm0311]

2차, 리뷰 의견과 질문입니다.

[uhm0311]

3차 리뷰 의견입니다.

[jhpark816]

@oliviarla @uhm0311
CompositeException의 printCallStack() 관련 PR은 merge 되었습니다.

[oliviarla]

@uhm0311 @jhpark816
현재 END/PIPE_ERROR가 오지 않아 예외가 발생한다면 아래와 같이 로깅이 되는데요, 이 때 어떤 에러가 어떤 연산에서 났는지 확인하기 어렵다는 단점이 있긴 합니다만, 별다른 해결 방안은 없는 것 같습니다. (pipe operation에서 내부적으로 사용하고 있는 index 정보를 같이 로깅하면 어떨까 생각해보았는데, 메서드 인자 등을 로그로 알기 어려우므로 index를 알더라도 쓸모없는 정보가 될 것 같습니다.)

OperationException: SERVER: Multiple exceptions (3) reported: SERVER_ERROR out of memory @ testnode 0.0.0.0/0.0.0.0:11211, CLIENT_ERROR too large value @ testnode 0.0.0.0/0.0.0.0:11211, END|PIPE_ERROR not received

[uhm0311]

4차 리뷰 의견입니다.

[oliviarla]

@jhpark816 리뷰 부탁드립니다.

[jhpark816]

@oliviarla 리뷰 진행 중입니다.

[jhpark816]

일부 리뷰

[jhpark816]

리뷰 완료,
복잡한 이슈네요.

[oliviarla]

@jhpark816 @uhm0311
현재 상황에서 수정할 부분을 대략 정리해봤습니다. 코드로 작성하기 전에 검토 먼저 부탁드립니다.

• OperationImpl#readFromBuffer의 로직 변경
  • isPipeOperation()가 true이고 첫 번째 응답이 RESPONSE <count> 라면, 이후에는 classifyError()를 거치지 않고 handleLine()에서
  • CLIENT_ERROR, SERVER_ERROR, ERROR를 읽고 gotStatus 콜백 메서드에 넘긴다.
• handleLine(String line)의 로직 변경
  • END가 왔을 때에는 Operation COMPLETE 상태로 변경
  • PIPE_ERROR가 왔을 때에는 Operation COMPLETE 상태로 변경하고, operation에 예외를 추가해 reconnect되도록 하고, hasErrored가 true를 반환하도록 한다.
• getFailedResult() 메서드를 pipe 관련 Future에 추가
  • 에러와 상관없이 failedResult 결과를 조회할 수 있도록 한다.

[jhpark816]

@oliviarla

isPipeOperation()가 true이면 classifyError()를 거치지 않고 handleLine()에서 CLIENT_ERROR, SERVER_ERROR, ERROR를 gotStatus 콜백 메서드에 넘긴다.

• pipe 요청에 대해 ERROR 등이 바로 나올 수 있으며, 이 경우는 classifyError()를 수행해야 합니다.
• pipe 요청에 대해 RESPONSE <count>\r\n 응답이 온 이후에는 classifyError()를 거치지 않아야 합니다.

END가 오면 리다이렉트한 후 Operation COMPLETE 상태로 변경
PIPE_ERROR가 오면 리다이렉트한 후 Operation COMPLETE 상태로 변경

리다이렉트한다는 것이 어떤 의미인가요?

future.get()의 로직 변경
hasErrored 일 때에는 exception을 던지지 않는다.

result가 있는 지에 따라 판단해야 할 지 ? 정도의 느낌이 듭니다.

[oliviarla]

@jhpark816
윗 코멘트를 수정했습니다.

[jhpark816]

hasErrored로 인한 exception은 던지지 않는다. cancel에 대한 exception은 던진다. (이 때에는 어디까지 처리되었는지 여부는 알 수 없다.)

• exception, cancel, timeout 관련 처리는 동일해야 할 것 같습니다.
• future.get() 동작을 아래와 같이 수정하는 방안이 있습니다. (backward compatibility 이슈 발생)
  • error, cancel, timeout 시에 exception throw 하지 않는 방식이 있습니다.
  • 이 경우, error, cacncel, timeout을 OperationStatus로 리턴해야 합니다.
• future.get() 동작을 그대로 유지하는 대신, getSome() 같은 별도의 메소드를 두는 방식이 있습니다.
  • getSome()은 error, cacncel, timeout을 OperationStatus로 리턴합니다.
  • getSome()은 getBulk에서 제공하는 메소드이고, storeBulk에 맞는 용어의 메소드(getFailed())이어야 합니다.

[uhm0311]

END가 왔을 때에는 Operation COMPLETE 상태로 변경

현재와 동일한 동작이란 거죠?

getFailedResult() 메서드를 pipe 관련 Future에 추가

해당 메소드의 리턴 타입은 어떻게 되나요?

[oliviarla]

@uhm0311

현재와 동일한 동작이란 거죠?

네 맞습니다.

getFailedResult() 메서드를 pipe 관련 Future에 추가 -> 해당 메소드의 리턴 타입은 어떻게 되나요?

failedResult 필드 타입과 동일하게 반환할 예정입니다.

---

@jhpark816 @uhm0311

• 현재 failedResult과 관련 없이 일단 END|PIPE_ERROR까지 읽도록만 수정해두었습니다. (handleError에서 바로 throw하지 않도록 변경)
• failedResult는 다른 PR로 수정할 예정이며, 아마 PIPE 동기 방식 PR 이후에 적용해도 될 것 같습니다.
• handleLine, handleRead 일관성 문제는 https://github.com/jam2in/arcus-works/issues/659 이슈에서 별도로 다룰 예정입니다.

[jhpark816]

리뷰 완료

[oliviarla]

@jhpark816
일단 CollectionPipedInsertOperationImpl 만 수정해두었습니다. 해당 클래스에 대한 리뷰가 끝나면 CollectionPipedUpdateOperationImpl, CollectionPipedExistOperationImpl, CollectionBulkInsertOperationImpl 클래스에 동일하게 적용하겠습니다.

[jhpark816]

@uhm0311 리뷰 바랍니다.

[uhm0311]

변경 내역을 다 따라가지 못해서 질문 남깁니다.

[oliviarla]

@uhm0311 리뷰 완료되셨을까요?

[uhm0311]

@oliviarla

진행 중입니다.

[jhpark816]

리뷰 완료

[jhpark816]

Exception 처리가 필요하니 handleRead()가 어색하네요.

@oliviarla
pipe 연산에서 handleError() 메소드를 override하거나
처음 제안해 주었던 handleError() 메소드를 수정하는 방안이 괜찮았다고 생각되네요.

아래 코드는 처음 제안해 준 방안과 거의 유사하며,
ReadType으로 LINE, DATA 외에 next line을 읽는 NEXT 타입을 두고
pipe 연산에서 처음 line을 읽은 후에 ReadType을 NEXT로 변경한다고 가정한 것입니다.

  protected void handleError(OperationErrorType eType, String line)
          throws IOException {
    getLogger().error("Error:  %s by %s", line, this);
    switch (eType) {
      case GENERAL:
      case SERVER:
        . . .
        break;
      case CLIENT:
        . . .
        break;
      default:
        assert false;
    }
    if (readType == OperationReadType.LINE) {
      transitionState(OperationState.COMPLETE);
      throw exception;
    }
  }

그리고, readFromBuffer()는 아래와 같이 변경할 수 있습니다.

• NEXT인 경우 LINE과 동일하게 처리
• COMPLETE이고 hasErrored이면 throw Exception 처리
  • 아니면 handleLine()에서 throw Exception 하게 수정

  public void readFromBuffer(ByteBuffer data) throws IOException {
    // Loop while there's data remaining to get it all drained.
    while (data.remaining() > 0) {
      if (getState() == OperationState.COMPLETE ||
          getState() == OperationState.MOVING || // ENABLE_REPLICATION
          getState() == OperationState.REDIRECT) { // ENABLE_MIGRATION
        break;
      }
      if (readType == OperationReadType.LINE ||
          readType == OperationReadType.NEXT) {
        String line = getLineFromBuffer(data);
        if (line == null) {
          continue;
        }
        OperationErrorType eType = classifyError(line);
        if (eType != null) {
          handleError(eType, line);
        } else { 
          handleLine(line);
        }
      } else { // OperationReadType.DATA
        handleRead(data);
      }
    }
    if (getState() == OperationState.COMPLETE && hasErrored()) {
      throw getException();
    }
  }

그러면, pipe 연산에서는 handleLine 메소드만 있으면 됩니다.
handleNOTPiped() 메소드를 따로 두어 분리하는 것은 읽기에 나은 것 같습니다.

이래저래 생각을 해 보니,
처음 제안해 주었던 방식이 나은 것 같아 위와 같이 의견을 전달합니다.
검토해 보세요.

[oliviarla]

handleError를 오버라이드하는 형태로 구현해봤습니다.
이를 통해 공통 Operation 처리 코드 중 readFromBuffer 쪽만 수정하도록 변경되었습니다.