DetectDuplicate · Wait · Notify · LookupRecord 같은 프로세서는 상태를 저장할 캐시를 요구한다. NiFi 가 기본으로 제공하는 것이 DistributedMapCacheServer 와 DistributedMapCacheClientService 두 컨트롤러 서비스다.
가장 흔한 실수는 클라이언트만 활성화하고 서버를 활성화하지 않는 것이다. 프로세서 속성에서 고르는 것은 클라이언트이므로 서버의 존재를 잊기 쉽다.
Failed to properly initialize Processor. If still scheduled to run, NiFi will attempt to
initialize and run the Processor again after the 'Administrative Yield Duration' has elapsed.
Failure is due to java.io.UncheckedIOException:
io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: localhost/127.0.0.1:4557
Connection refused 는 서버가 그 포트를 듣고 있지 않다는 뜻이다. 클라이언트 설정이 아니라 서버 쪽을 먼저 본다.
별도 프로세스를 띄울 필요는 없다. 서버는 NiFi JVM 안에서 컨트롤러 서비스로 돌면서 포트를 직접 연다.
| 서비스 | 역할 | 핵심 속성 |
|---|---|---|
DistributedMapCacheServer |
캐시 저장소. 포트를 연다 | Port(기본 4557), Maximum Cache Entries, Eviction Strategy, Persistence Directory |
DistributedMapCacheClientService |
프로세서가 쓰는 접속 창구 | Server Hostname, Server Port |
순서는 서버를 먼저 Enable 하고, 포트가 열린 것을 확인한 뒤 클라이언트를 Enable 한다.
ss -tnlp | grep 4557
grep -i DistributedMapCacheServer ${NIFI_HOME}/logs/nifi-app.log | tail -20
서버가 정상 기동하면 로그에 기동 메시지가 남는다. 아무것도 없으면 Enable 자체가 실패한 것이므로, NiFi UI 의 컨트롤러 서비스 목록에서 해당 항목에 마우스를 올려 오류 문구를 읽는다. 포트 충돌(Address already in use)이 가장 흔하다.
서버 서비스는 각 노드에서 각자 인스턴스를 띄운다. 따라서 클라이언트가 localhost 를 보면 노드마다 자기 자신의 캐시를 쓰게 되고, 캐시가 노드 수만큼 쪼개진다. DetectDuplicate 로 중복을 걸러 내려는 흐름에서는 이것이 곧 중복 통과를 뜻한다.
클러스터에서 상태를 공유하려면 클라이언트의 Server Hostname 을 특정 노드의 FQDN 하나로 고정한다.
Server Hostname : nifi01.example.local
Server Port : 4557
그 노드가 단일 장애점이 되므로, 정지했을 때 흐름이 어떻게 되는지를 미리 정해 둔다. 실패를 그대로 흘려보내도 되는 용도(중복 제거의 정확도가 절대적이지 않은 경우)라면 감수할 만하고, 그렇지 않다면 외부 저장소를 쓰는 편이 낫다.
고정한 노드의 방화벽에서 다른 노드로부터 오는 4557 을 열어야 한다.
nc -zv nifi01.example.local 4557
캐시가 흐름의 정확성에 직결된다면 RedisDistributedMapCacheClientService 또는 HBase_2_ClientMapCacheService 를 쓴다. NiFi 재시작이나 노드 교체와 무관하게 상태가 유지되고, 단일 노드 의존이 사라진다.
| 구현 | 특징 |
|---|---|
DistributedMapCacheServer |
설정이 가장 간단. NiFi 프로세스에 묶임 |
| Redis | 외부 의존이 생기지만 영속성과 가용성 확보 |
| HBase | 이미 HBase 를 쓰는 환경에서 추가 구성 요소 없이 |
DistributedMapCacheServer 에도 Persistence Directory 를 지정하면 재시작 후 내용이 복구되지만, 노드 간 공유 문제는 그대로 남는다.
캐시는 무한하지 않다. Maximum Cache Entries 를 넘으면 Eviction Strategy(LFU · LRU · FIFO)에 따라 오래된 항목이 빠진다. 중복 판정 기간이 길어야 한다면 그 기간 동안 들어올 키 수를 계산해 넉넉히 잡는다. DetectDuplicate 쪽에도 Age Off Duration 이 있으므로 두 값을 함께 맞춘다.
Cache Entry Identifier 는 키다. 파일 이름처럼 매번 달라지는 값을 쓰면 중복 판정이 되지 않는다. 내용 해시나 업무 키를 쓴다.