TableView에서 message 전체를 쓰고 싶을 때

Pulsar의 TableView는 topic을 key마다 최신 값 하나씩만 남긴 표처럼 보여 줍니다. 설정값이나 상태처럼 “지금 값"만 필요한 자료를 읽을 때 편합니다.

그런데 TableView가 돌려주는 것은 message의 value뿐이었습니다. message에 함께 실려 오는 properties나 event time은 볼 방법이 없었습니다. 이 요청은 apache/pulsar#24744에 올라와 있었고 설계는 PIP-445로 정리했습니다. 이 글에서는 이 빈자리를 어떤 모양의 API로 채웠는지, 그리고 왜 그 모양을 골랐는지 적습니다.

처음 떠올린 방법과 그 문제

가장 먼저 떠오르는 방법은 TableView에 “원본 message를 돌려주는 method"를 더하는 것입니다.

원본 message를 언제든 돌려주려면 TableView가 모든 message를 통째로 쥐고 있어야 한다는 문제가 있었습니다. 그러면 value만 필요한 기존 사용자까지 메모리를 더 쓰게 됩니다. 쓰지 않는 기능의 비용을 모두가 내는 셈입니다.

값을 만드는 방법을 사용자에게 맡겼습니다

그래서 방향을 바꿔 TableView가 무엇을 보관할지를 사용자가 정하게 했습니다. mapper를 넘기는 방식입니다. mapper는 message 하나를 받아서 보관할 값 하나로 바꾸는 함수입니다.

TableView<String> regions = client.newTableView(Schema.STRING)
        .topic("my-topic")
        .createMapped(msg -> msg.getProperty("region"));

property 하나만 필요하면 그것만 꺼내서 보관합니다. message 전체가 필요하면 그대로 돌려주면 됩니다.

.createMapped(msg -> msg)

TableViewMessageMapper의 모양은 단순합니다. 값을 만드는 method 하나와, 실패했을 때 불리는 method 하나입니다.

@FunctionalInterface
public interface TableViewMessageMapper<T, V> {

    // message 하나를 보관할 값 하나로 바꾼다
    V map(Message<T> message) throws Exception;

    // map이 실패하면 불린다
    default boolean onMappingError(Message<T> message, Throwable error) {
        // false면 기본 error log를 남긴다
        return false;
    }
}

기존의 create()는 그대로라서 기존 사용자의 비용은 늘지 않습니다. 필요한 사람만 mapper를 넘겨 필요한 만큼 가져가면 됩니다.

사용자 코드가 실패하면 어떻게 할 것인가

mapper는 사용자가 쓴 코드입니다. TableView 안에서 사용자 코드가 돌기 시작하면 그 코드가 예외를 던질 때 무슨 일이 일어나는지를 정해야 합니다. 이 부분에 가장 많은 시간을 썼습니다.

정한 규칙은 다음과 같습니다.

예외가 났을 때는 mapper의 onMappingError가 불립니다. 기본 동작은 topic, key, message id와 함께 error log를 남기는 것입니다. 직접 metric으로 세거나 다른 방식으로 알리고 싶으면 이 method를 재정의하면 됩니다.

실제로 처리하는 곳은 AbstractTableViewImpl.java의 handleKeyedMessage()입니다. 예외를 잡아서 그 message만 건너뜁니다.

try {
    // 이 안에서 mapper가 불린다
    cur = getValueIfPresent(msg);
} catch (Throwable t) {
    /*
     * 읽은 위치는 앞으로 옮긴다.
     * 그래야 refresh()가 이 message에서 멈추지 않는다.
     */
    lastReadPositions.put(msg.getTopicName(), msg.getMessageId());
    // false가 오면 기본 log를 남긴다
    if (!onMappingError(msg, t)) {
        log.error().attr("key", key)
                .attr("messageId", msg.getMessageId())
                .exception(t)
                .log("Skipping message whose value could not be decoded or mapped");
    }
    // key의 값은 그대로 두고 다음 message로
    return;
}

한 message의 실패가 TableView 전체를 멈추게 하지는 않습니다.

같이 쓸 수 없는 조합은 미리 막았습니다

mapper를 쓰는 view에서는 topicCompactionStrategyClassName 설정을 받지 않습니다. 이 strategy는 topic의 원래 value type끼리 비교하는데, mapper를 거친 view는 다른 type의 값을 보관하기 때문입니다. 그대로 두면 실행 도중에 ClassCastException이 납니다. 그래서 만들 때 바로 거절하도록 했습니다.

또 mapper가 message를 그대로 붙잡고 있을 수 있으므로 mapper를 쓰는 view에서는 message 재사용(pooling)을 끕니다. msg -> msg처럼 message 전체를 보관하면 metadata와 schema까지 함께 붙잡게 됩니다. 그래서 key가 많은 topic이라면 필요한 값만 담은 객체를 만들어 보관하는 편이 낫습니다.

하는 김에 드러난 기존 문제

이 작업을 하면서 기존 동작의 빈틈도 드러났습니다. value를 읽다가 예외가 나면 처음 내용을 읽어 들이는 동안에는 TableView 생성 자체가 실패하면서 reader가 닫히지 않고 남았습니다. 그 뒤 새 message를 따라 읽는 동안에는 message가 조용히 버려졌습니다.

이제는 두 경우 모두 같은 방식으로 건너뛰고 log를 남깁니다. 시작에 실패하면 reader도 닫습니다.

마치며

처음에는 method 하나를 더하면 끝날 일처럼 보였습니다. 그런데 막상 해 보니 “누가 비용을 내는가"와 “사용자 코드가 실패하면 어떻게 되는가"를 정하는 일이 대부분이었습니다.

공개 API에 사용자 코드를 받아들이는 자리를 만들 때는 그 코드가 실패했을 때의 규칙부터 문서에 적고 시작해야 한다고 배웠습니다.