executor를 닫으면 대기 중이던 작업은 어떻게 될까

다른 변경을 올리고 CI를 돌리던 중에 제 변경과 상관없는 테스트 하나가 시간 초과로 실패했습니다. 이 실패는 apache/pulsar#26685에 정리해 두었습니다. 다시 돌려도 또 실패했습니다. client를 닫는 동안 계속 message를 보내는 테스트였습니다. 보낸 message 가운데 일부가 성공도 실패도 아닌 채로 영원히 끝나지 않고 있었습니다.

이 글에서는 그 원인이었던 shutdownNow()의 동작과 그래서 생긴 “끝나지 않는 future"를 어떻게 고쳤는지 적습니다.

결과를 알려 주는 일은 다른 thread가 맡습니다

Pulsar의 새 client(v5)에서 message를 보내면 CompletableFuture를 돌려받습니다. 전송이 끝나면 이 future가 성공이나 실패로 완료됩니다.

future를 완료하면 거기에 걸어 둔 callback이 그 자리에서 실행됩니다. 사용자가 쓴 callback이 네트워크를 처리하는 thread에서 오래 돌면 곤란합니다. 그래서 producer는 future를 완료하는 일 자체를 client가 가진 별도의 executor에 넘깁니다.

평소에는 아무 문제가 없다가 client가 닫히는 순간에 문제가 생깁니다.

shutdown()과 shutdownNow()는 다릅니다

Java의 ExecutorService를 닫는 방법은 두 가지입니다.

shutdownNow()는 대기 중이던 작업을 실행하지 않습니다. 아래 코드로 확인할 수 있습니다.

ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<String> future = new CompletableFuture<>();

// thread를 붙잡아 둔다
executor.execute(() -> sleep(1000));
// 대기열에서 기다린다
executor.execute(() -> future.complete("done"));

// 기다리던 작업은 버려진다
executor.shutdownNow();
// TimeoutException
future.get(5, TimeUnit.SECONDS);

두 번째 작업은 영영 실행되지 않고 future는 누구도 완료해 주지 않습니다. 예외도 나지 않습니다. 기다리는 쪽은 그냥 계속 기다립니다.

Pulsar에서 벌어진 일

Pulsar client는 닫힐 때 자기 executor를 shutdownNow()로 정리합니다. PulsarClientImpl.java의 shutdownExecutors()를 보면 됩니다.

private void shutdownExecutors() throws PulsarClientException {
    PulsarClientException pulsarClientException = null;
    if (createdExecutorProviders) {

        if (externalExecutorProvider != null && !externalExecutorProvider.isShutdown()) {
            try {
                // 남은 작업은 실행하지 않고 버린다
                externalExecutorProvider.shutdownNow();

producer가 결과를 넘기는 executor가 바로 이 externalExecutorProvider에서 나옵니다. 그래서 다음 순서가 가능했습니다.

1. client가 닫히기 시작합니다.
2. 그 사이에 보낸 message가 “client가 닫히는 중"이라는 이유로 실패합니다.
3. producer가 “이 future를 실패로 완료하라"는 작업을 executor에 넘깁니다.
4. 그 작업이 실행되기 전에 executor가 shutdownNow()로 닫힙니다.
5. 작업은 버려지고 future는 끝나지 않습니다.

기존 코드에도 대비책이 하나 있었습니다. executor가 이미 닫혀서 작업을 받아 주지 않으면(RejectedExecutionException) 그 자리에서 직접 완료하도록 되어 있었습니다. 하지만 이 대비책은 executor가 완전히 닫힌 뒤에만 통합니다. 닫히기 직전에 대기열에 들어간 작업은 지켜 주지 못합니다.

작업을 받아 주지 않을 때는 예외라도 나지만 받아 놓고 버린 작업은 아무 신호도 남기지 않습니다.

닫히는 중이면 넘기지 않고 그 자리에서 끝냅니다

고친 곳은 ScalableTopicProducer.java의 finish()이고 내용은 조건 하나입니다. client가 닫히는 중이면 executor에 넘기지 않고 실패를 알아챈 thread에서 바로 future를 완료합니다.

 boolean offload = send.dispatching || (failure != null && !isEventLoopThread());
-if (!offload) {
+/*
+ * 더한 조건.
+ * client가 닫히는 중이면 executor에 넘기지 않고,
+ * 지금 이 thread에서 바로 완료한다.
+ */
+if (!offload || client.v4Client().isClosed()) {
     complete(send, messageId, failure);
     return;
 }
 try {
     // 평소에는 executor에 넘긴다
     completionExecutor.execute(() -> complete(send, messageId, failure));
 } catch (RejectedExecutionException e) {
     /*
      * 기존 대비책.
      * executor가 이미 닫혀서 작업을 받아 주지 않을 때만 통한다.
      */
     complete(send, messageId, failure);
 }

client가 열려 있는 동안의 동작은 그대로입니다. 결과는 여전히 별도의 executor에서 전달됩니다.

고쳐졌는지 어떻게 확인했나

이 문제는 타이밍이 맞아야만 생깁니다. 그래서 우연에 기대지 않고 그 타이밍을 직접 만들도록 테스트를 썼습니다.

1. client의 completion thread를 하나로 줄이고 그 하나를 일부러 붙잡아 둡니다.
2. 다른 thread에서 client를 닫기 시작합니다.
3. client가 닫히는 중이 되면 message를 하나 보냅니다.
4. 그 send가 끝나지 않는 것이 아니라 AlreadyClosedException으로 실패하는지 확인합니다.

마치며

shutdownNow()가 대기 중인 작업을 버리는 건 문서에도 적혀 있는 동작입니다. 다만 그 작업이 누군가 기다리고 있는 future를 완료하는 일이라면 버려지는 순간 기다리는 쪽은 영원히 멈춥니다.

future를 다른 thread에서 완료하도록 넘긴다면 그 thread가 사라질 때 누가 대신 완료해 줄지도 함께 정해 두어야 합니다. 이번에 그걸 배웠습니다.