Свойство типа Controller Service в кастомном процессоре NiFi

Преамбула

В одном из самодельных процессоров Apache NiFi у меня возникла необходимость работать с файлами сертификатов. Так как в проекте уже были настроенные службы сертификатов (типа StandardRestrictedSSLContextService), я решил сделать у процессора свойство с типом SSLContextService, чтобы можно было подставлять уже настроенные в NiFi службы контроллера и брать данные сертификатов оттуда. Изучил матчасть, ничего сложного. Добавил свойство в процессор, закинул его на flow. Но… процессор не видит мои StandardRestrictedSSLContextService. Пересмотрел и перечитал множество статей, в том числе от уважаемого Pierre Villard, но никак. Пока не наткнулся на реализацию похожего кейса на Github.

Возможно, это банальная проблема, но у меня она вызвала приличные затруднения, поэтому я решил об этом написать. Ниже приведу примеры кода, а так же покажу, как подсунуть свою заглушку в интеграционный тест процессора (что тоже оказалось не совсем очевидной задачей).

Добавление свойства в процессор

Добавляем в процессор свойство, через которое можно будет указывать стандартные службы, обеспечивающие доступ к сертификатам.

public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()     .name("SSL Context Service")     .description("SSL Context Service provides trusted certificates and client certificates for TLS communication.")     .required(false)     .identifiesControllerService(SSLContextService.class)     .build();

В коде метода @OnTrigger процессора доступ к значению свойства будем получать вот так:

SSLContextService sslService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);

Теперь наверно самый важный участок этой маленькой статьи, ради которого все и затевалось. Правильное добавление зависимостей!

1) В pom.xml основного модуля процессора добавим две зависимости:

<dependency>     <groupId>org.apache.nifi</groupId>     <artifactId>nifi-ssl-context-service-api</artifactId>     <version>${nifi.version}</version> </dependency>   <dependency>     <groupId>org.apache.nifi</groupId>     <artifactId>nifi-standard-services-api-nar</artifactId>     <version>${nifi.version}</version>     <type>nar</type> </dependency>

2) В pom.xml nar-модуля добавим одну зависимость:

<dependency>     <groupId>org.apache.nifi</groupId>     <artifactId>nifi-standard-services-api-nar</artifactId>     <version>${nifi.version}</version>     <type>nar</type> </dependency>

nifi-ssl-context-service-api — с помощью этой зависимости в модуле процессора у нас появляется возможность использовать интерфейс SSLContextService.class в коде основного модуля процессора

nifi-standard-services-api-nar — эта зависимость в обоих модулях позволит процессору использовать стандартные службы NiFi

Собственно, вот и весь фокус.

Если надо понять, какую зависимость придется использовать, когда нужно добавить в процессор свойство, отличное по типу от SSLContextService, то помочь в этом может анализ исходников NiFi.

Тестирование процессора, в котором используется служба в качестве одного из свойств

Небольшая проблема заключается в том, что для тестирования процессора, у которого одно из свойств это Controller Service, нам надо обязательно создать некую заглушку этого сервиса. А потом и указать ее в качестве одного из свойств.

Подготовим вспомогательные тестовые данные:

import javax.net.ssl.SSLContext; import javax.net.ssl.X509TrustManager; import lombok.Builder; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.controller.AbstractControllerService; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.security.util.TlsConfiguration; import org.apache.nifi.ssl.SSLContextService;  @Builder public class DummySSLContextService extends AbstractControllerService implements SSLContextService {     private final String keyStoreFile;     private final String keyStorePassword;     private final String keyStoreType;     private final String trustStoreFile;     private final String trustStorePassword;     private final String trustStoreType;      @Override     public TlsConfiguration createTlsConfiguration() {         throw new RuntimeException("Method not implemented");     }      @Override     public SSLContext createContext() {         throw new RuntimeException("Method not implemented");     }      @Override     public SSLContext createSSLContext(org.apache.nifi.security.util.ClientAuth clientAuth) throws ProcessException {         throw new RuntimeException("Method not implemented");     }      @Override     public SSLContext createSSLContext(ClientAuth clientAuth) throws ProcessException {         throw new RuntimeException("Method not implemented");     }      @Override     public X509TrustManager createTrustManager() {         throw new RuntimeException("Method not implemented");     }      @Override     public String getTrustStoreFile() {         return trustStoreFile;     }      @Override     public String getTrustStoreType() {         return trustStoreType;     }      @Override     public String getTrustStorePassword() {         return trustStorePassword;     }      @Override     public boolean isTrustStoreConfigured() {         throw new RuntimeException("Method not implemented");     }      @Override     public String getKeyStoreFile() {         return keyStoreFile;     }      @Override     public String getKeyStoreType() {         return keyStoreType;     }      @Override     public String getKeyStorePassword() {         return keyStorePassword;     }      @Override     public String getKeyPassword() {         throw new RuntimeException("Method not implemented");     }      @Override     public boolean isKeyStoreConfigured() {         throw new RuntimeException("Method not implemented");     }      @Override     public String getSslAlgorithm() {         throw new RuntimeException("Method not implemented");     }      @Override     public void onPropertyModified(PropertyDescriptor propertyDescriptor, String s, String s1) {         // ничего не делаем     } }

import java.nio.file.Path;  import org.apache.nifi.ssl.SSLContextService;  public class TestData {     static final String CLIENT_CERT_FILE_NAME = Path.of("src/test/resources/client-identity.jks").toAbsolutePath().toString();     static final String CLIENT_TRUST_FILE_NAME = Path.of("src/test/resources/truststore.jks").toAbsolutePath().toString();     public static final String KEYSTORE_COMMON_PASSWORD = "changeit";     public static final String KEYSTORE_COMMON_TYPE = "JKS";      public static final SSLContextService TEST_SSL_CONTEXT_SERVICE = DummySSLContextService.builder()             .keyStoreFile(CLIENT_CERT_FILE_NAME)             .keyStorePassword(KEYSTORE_COMMON_PASSWORD)             .keyStoreType(KEYSTORE_COMMON_TYPE)             .trustStoreFile(CLIENT_TRUST_FILE_NAME)             .trustStorePassword(KEYSTORE_COMMON_PASSWORD)             .trustStoreType(KEYSTORE_COMMON_TYPE)             .build(); }

И собственно тест (тут SSL_CONTEXT_SERVICE это дескриптор свойства процессора):

    @Test     void should_run_with_service_property() throws InitializationException {         final TestRunner testRunner = TestRunners.newTestRunner(MyProcessor.class);          // устанавливаем значение для одного из свойств тестового процессора         testRunner.setProperty(SOME_PROPERTY_NAME, SOME_PROPERTY_VALUE);                // устанавливаем для тестового процессора свойство, в котором указывается Controller Service         // это делается в три шага         testRunner.addControllerService("TestSSLContextService", TEST_SSL_CONTEXT_SERVICE);         testRunner.setProperty(SSL_CONTEXT_SERVICE, "TestSSLContextService");         testRunner.enableControllerService(TEST_SSL_CONTEXT_SERVICE);          // создаем набор аттрибутов для тестового flow-файла         Map<String, String> attributes = new HashMap<>();         attributes.put(SOME_ATTR_NAME, SOME_ATTR_VALUE);          // добавляем контент и аттрибуты в тестовый flow-файл         // вместо строки сюда может быть передан поток         testRunner.enqueue("Flowfile content", attributes);          // When         testRunner.run();          // Then         List<MockFlowFile> originalFlowFiles = testRunner.getFlowFilesForRelationship(REL_SUCCESS);         List<MockFlowFile> failureFlowFiles = testRunner.getFlowFilesForRelationship(REL_FAILURE);          assertThat(originalFlowFiles.size()).isEqualTo(1);         assertThat(failureFlowFiles.size()).isZero();          Map<String, String> actualAttributes = originalFlowFiles.get(0).getAttributes();         assertThat(actualAttributes)                 .isNotNull()                 .containsEntry(SOME_ATTR_NAME, SOME_ATTR_VALUE);     }

Заключение

Вот и все. Надеюсь эта маленькая статья будет полезна. Замечания и указания на неточности в комментариях очень приветствуются))

Спасибо, что дочитали до конца.