Pulsar 플러그인 개발

Pulsar 플러그인 개발 (Pulsar Plugin Development)

Pulsar는 엔트리 필터, 프로토콜 핸들러, 인터셉터처럼 다양한 확장점을 플러그인으로 개발할 수 있게 해줘요. 이 글에서는 그중 추가 서블릿(additional servlet)과 엔트리 필터(entry filter)를 Maven 프로젝트로 만들고 NAR 파일로 패키징해 브로커에 연결하는 전체 과정을 정리해 드릴게요.

출처: 문서

본문

Pulsar를 위해 엔트리 필터, 프로토콜 핸들러, 인터셉터 등 다양한 플러그인을 개발할 수 있어요.

추가 서블릿 (Additional Servlets)

이 장에서는 추가 서블릿이 무엇인지와 사용 방법을 설명해요.

추가 서블릿이 무엇인가요?

Pulsar는 상호 작용을 위한 다양한 REST API를 제공해요. 추가 커스텀 로직을 REST API로 노출하기 위해 Pulsar는 추가 서블릿(additional servlet) 개념을 제공해요. 이 서블릿은 broker 또는 pulsar proxy에서 플러그인으로 실행돼요.

추가 서블릿은 어떻게 사용하나요?

이 예시 구현을 살펴보거나 아래 단계를 따르세요:

1. Maven 프로젝트를 만들어요.
2. AdditionalServlet 또는 AdditionalServletWithPulsarService 인터페이스를 구현해요.
3. 프로젝트를 NAR 파일로 패키징해요.
4. broker.conf 파일(또는 standalone.conf 파일)을 구성하고 브로커를 재시작해요.

1단계: Maven 프로젝트 만들기

Maven 프로젝트를 만드는 방법은 여기를 보세요.

2단계: AdditionalServlet 인터페이스 구현

1. 표시된 대로 pom.xml 파일에 pulsar-broker 의존성을 추가해요. 그렇지 않으면 AdditionalServlet 인터페이스를 찾을 수 없어요.

   <dependency>
      <groupId>org.apache.pulsar</groupId>
      <artifactId>pulsar-broker</artifactId>
      <version>${pulsar.version}</version>
      <scope>provided</scope>
   </dependency>
   

2. AdditionalServlet 인터페이스의 메서드를 구현해요.

   • loadConfig는 PulsarConfiguration에서 구성 속성을 로드해 서블릿을 구성할 수 있게 해줘요.
   • getBasePath는 서블릿이 로드될 경로를 정의해요.
   • getServletHolder는 이 서블릿의 ServletHolder를 반환해요.
   • close는 리소스를 해제할 수 있게 해줘요.

3. NAR 파일을 설명해요. NAR 파일을 설명하려면 resources/META-INF/services 디렉터리에 additional_servlet.yml 파일을 만들어요.

   name: my-servlet
   description: Describes my-servlet
   additionalServletClass: org.my.package.MyServlet
   

3단계: 프로젝트를 NAR 파일로 패키징

1. NAR 파일의 컴파일된 플러그인을 pom.xml 파일에 추가해요.

   <build>
      <finalName>${project.artifactId}</finalName>
      <plugins>
         <plugin>
            <groupId>org.apache.nifi</groupId>
            <artifactId>nifi-nar-maven-plugin</artifactId>
            <version>1.5.0</version>
            <extensions>true</extensions>
            <configuration>
               <finalName>${project.artifactId}-${project.version}</finalName>
            </configuration>
            <executions>
               <execution>
                  <id>default-nar</id>
                  <phase>package</phase>
                  <goals>
                     <goal>nar</goal>
                  </goals>
               </execution>
            </executions>
         </plugin>
      </plugins>
   </build>
   

2. target 디렉터리에서 NAR 파일을 생성해요.

   mvn clean install
   

4단계: 브로커 구성 및 재시작

1. broker.conf 파일(또는 standalone.conf 파일)에서 다음 파라미터를 구성해요.

   # Name of pluggable additional servlets
   # Multiple servlets need to be separated by commas.
   additionalServlets=my-servlet
   # The directory for all additional servlet implementations
   additionalServletDirectory=tempDir
   

2. 브로커를 재시작해요. 플러그인이 성공적으로 로드되면 다음 브로커 로그를 볼 수 있어요.

   Successfully loaded additional servlet for name `my-servlet`
   

엔트리 필터 (Entry Filter)

이 장에서는 엔트리 필터가 무엇인지 설명하고 엔트리 필터 사용 방법을 보여줘요.

엔트리 필터가 무엇인가요?

엔트리 필터는 커스텀 메시지 엔트리 전략을 구현하기 위한 확장 지점이에요. 엔트리 필터로 소비자에게 메시지를 보낼지 여부(브로커는 엔트리 필터의 반환값을 사용해 메시지를 보내야 할지 버려야 할지 결정할 수 있어요) 또는 특정 소비자에게 메시지를 보낼지를 결정할 수 있어요.

태그된 메시지나 커스텀 지연 메시지 같은 기능을 구현하려면 subscriptionProperties, properties과 엔트리 필터를 사용해요.

엔트리 필터는 어떻게 사용하나요?

아래 단계를 따르세요:

1. Maven 프로젝트를 만들어요.
2. EntryFilter 인터페이스를 구현해요.
3. 구현 클래스를 NAR 파일로 패키징해요.
4. broker.conf 파일(또는 standalone.conf 파일)을 구성하고 브로커를 재시작해요.

1단계: Maven 프로젝트 만들기

Maven 프로젝트를 만드는 방법은 여기를 보세요.

2단계: EntryFilter 인터페이스 구현

1. 표시하기 위해 pom.xml 파일에 Pulsar broker 의존성을 추가해요. 그렇지 않으면 EntryFilter 인터페이스를 찾을 수 없어요.

   <dependency>
   <groupId>org.apache.pulsar</groupId>
   <artifactId>pulsar-broker</artifactId>
   <version>${pulsar.version}</version>
   <scope>provided</scope>
   </dependency>
   

2. FilterResult filterEntry(Entry entry, FilterContext context); 메서드를 구현해요. FilterContext를 통해 엔트리 메타데이터, 구독 및 기타 정보를 얻을 수 있어요.

   • 메서드가 ACCEPT 또는 NULL을 반환하면 이 메시지는 소비자에게 전송돼요.
   • 메서드가 REJECT를 반환하면 이 메시지는 필터링되고 메시지 퍼밋을 소비하지 않아요.
   • 여러 엔트리 필터가 있으면 이 메시지는 파이프라인의 모든 필터를 라운드로빈 방식으로 통과해요. 어떤 엔트리 필터가 REJECT를 반환하면 이 메시지는 버려져요.

3. NAR 파일을 설명해요. NAR 파일을 설명하려면 resources/META-INF/services 디렉터리에 entry_filter.yml 또는 entry_filter.yaml 파일을 만들어요.

   # Entry filter name, which should be configured in the broker.conf file later
   name: entryFilter
   # Entry filter description
   description: entry filter
   # Implementation class name of entry filter
   entryFilterClass: com.xxxx.xxxx.xxxx.DefaultEntryFilterImpl
   

3단계: 엔트리 필터 구현 클래스를 NAR 파일로 패키징

1. NAR 파일의 컴파일된 플러그인을 pom.xml 파일에 추가해요.

   <build>
           <finalName>${project.artifactId}</finalName>
           <plugins>
               <plugin>
                   <groupId>org.apache.nifi</groupId>
                   <artifactId>nifi-nar-maven-plugin</artifactId>
                   <version>1.5.0</version>
                   <extensions>true</extensions>
                   <configuration>
                       <finalName>${project.artifactId}-${project.version}</finalName>
                   </configuration>
                   <executions>
                       <execution>
                           <id>default-nar</id>
                           <phase>package</phase>
                           <goals>
                               <goal>nar</goal>
                           </goals>
                       </execution>
                   </executions>
               </plugin>
           </plugins>
       </build>
   

2. target 디렉터리에서 NAR 파일을 생성해요.

   mvn clean install
   

4단계: 브로커 구성 및 재시작

1. broker.conf 파일(또는 standalone.conf 파일)에서 다음 파라미터를 구성해요.

   # Class name of pluggable entry filters
   # Multiple classes need to be separated by commas.
   entryFilterNames=entryFilter1,entryFilter2,entryFilter3
   # The directory for all entry filter implementations
   entryFiltersDirectory=tempDir
   

2. 브로커를 재시작해요. 플러그인이 성공적으로 로드되면 다음 브로커 로그를 볼 수 있어요.

   Successfully loaded entry filter for name `{name of your entry filter}`
   

더 알아보기 (Learn more)

• 개발자를 위한 Pulsar — 다른 개발 문서를 살펴봐요.
• 브로커 구성 — 플러그인 관련 broker.conf 파라미터를 확인해요.
• NAR 패키징 (NiFi) — NAR 파일 구조를 이해해요.