@Test public void testCustomValidate() throws InitializationException, IOException { final TestRunner runner = TestRunners.newTestRunner(TestProcessor.class); final String tableName = "nifi"; final Table table = Mockito.mock(Table.class); when(table.getName()).thenReturn(TableName.valueOf(tableName)); MockHBaseClientService service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.enableControllerService(service); runner.assertNotValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.HADOOP_CONF_FILES, "src/test/resources/hbase-site.xml"); runner.enableControllerService(service); runner.assertValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_QUORUM, "localhost"); runner.enableControllerService(service); runner.assertNotValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_QUORUM, "localhost"); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_CLIENT_PORT, "2181"); runner.enableControllerService(service); runner.assertNotValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_QUORUM, "localhost"); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_CLIENT_PORT, "2181"); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_ZNODE_PARENT, "/hbase"); runner.enableControllerService(service); runner.assertValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.HADOOP_CONF_FILES, "src/test/resources/hbase-site.xml"); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_QUORUM, "localhost"); runner.setProperty(service, HBase_1_1_2_ClientService.ZOOKEEPER_CLIENT_PORT, "2181"); runner.enableControllerService(service); runner.assertValid(service); runner.removeControllerService(service); service = new MockHBaseClientService(table, kerberosPropsWithFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.HADOOP_CONF_FILES, "src/test/resources/hbase-site-security.xml"); runner.setProperty(service, kerberosPropsWithFile.getKerberosPrincipal(), "test@REALM"); runner.enableControllerService(service); runner.assertValid(service); runner.disableControllerService(service); runner.setProperty(service, HBase_1_1_2_ClientService.HADOOP_CONF_FILES, "src/test/resources/hbase-site-security.xml, src/test/resources/core-site-security.xml"); runner.enableControllerService(service); runner.assertNotValid(service); runner.disableControllerService(service); runner.setProperty(service, kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/fake.keytab"); runner.setProperty(service, kerberosPropsWithFile.getKerberosPrincipal(), "test@REALM"); runner.enableControllerService(service); runner.assertValid(service); runner.disableControllerService(service); runner.setProperty(service, kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/missing.keytab"); runner.enableControllerService(service); runner.assertNotValid(service); runner.disableControllerService(service); runner.setProperty(service, kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/fake.keytab"); runner.setProperty(service, kerberosPropsWithFile.getKerberosPrincipal(), ""); runner.enableControllerService(service); runner.assertNotValid(service); service = new MockHBaseClientService(table, kerberosPropsWithoutFile); runner.addControllerService("hbaseClientService", service); runner.setProperty(service, HBase_1_1_2_ClientService.HADOOP_CONF_FILES, "src/test/resources/hbase-site-security.xml, src/test/resources/core-site-security.xml"); runner.setProperty(service, kerberosPropsWithoutFile.getKerberosKeytab(), "src/test/resources/fake.keytab"); runner.setProperty(service, kerberosPropsWithoutFile.getKerberosPrincipal(), "test@REALM"); runner.enableControllerService(service); runner.assertNotValid(service); }
@Override protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { boolean confFileProvided = validationContext.getProperty(HADOOP_CONF_FILES).isSet(); boolean zkQuorumProvided = validationContext.getProperty(ZOOKEEPER_QUORUM).isSet(); boolean zkPortProvided = validationContext.getProperty(ZOOKEEPER_CLIENT_PORT).isSet(); boolean znodeParentProvided = validationContext.getProperty(ZOOKEEPER_ZNODE_PARENT).isSet(); boolean retriesProvided = validationContext.getProperty(HBASE_CLIENT_RETRIES).isSet(); final List<ValidationResult> problems = new ArrayList<>(); if (!confFileProvided && (!zkQuorumProvided || !zkPortProvided || !znodeParentProvided || !retriesProvided)) { problems.add(new ValidationResult.Builder() .valid(false) .subject(this.getClass().getSimpleName()) .explanation("ZooKeeper Quorum, ZooKeeper Client Port, ZooKeeper ZNode Parent, and HBase Client Retries are required " + "when Hadoop Configuration Files are not provided.") .build()); } if (confFileProvided) { final String configFiles = validationContext.getProperty(HADOOP_CONF_FILES).getValue(); ValidationResources resources = validationResourceHolder.get(); if (resources == null || !configFiles.equals(resources.getConfigResources())) { getLogger().debug("Reloading validation resources"); resources = new ValidationResources(configFiles, getConfigurationFromFiles(configFiles)); validationResourceHolder.set(resources); } final Configuration hbaseConfig = resources.getConfiguration(); final String principal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); problems.addAll(KerberosProperties.validatePrincipalAndKeytab( this.getClass().getSimpleName(), hbaseConfig, principal, keytab, getLogger())); } return problems; }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { boolean confFileProvided = validationContext.getProperty(HADOOP_CONF_FILES).isSet(); boolean zkQuorumProvided = validationContext.getProperty(ZOOKEEPER_QUORUM).isSet(); boolean zkPortProvided = validationContext.getProperty(ZOOKEEPER_CLIENT_PORT).isSet(); boolean znodeParentProvided = validationContext.getProperty(ZOOKEEPER_ZNODE_PARENT).isSet(); boolean retriesProvided = validationContext.getProperty(HBASE_CLIENT_RETRIES).isSet(); final List<ValidationResult> problems = new ArrayList<>(); if (!confFileProvided && (!zkQuorumProvided || !zkPortProvided || !znodeParentProvided || !retriesProvided)) { problems.add(new ValidationResult.Builder() .valid(false) .subject(this.getClass().getSimpleName()) .explanation("ZooKeeper Quorum, ZooKeeper Client Port, ZooKeeper ZNode Parent, and HBase Client Retries are required " + "when Hadoop Configuration Files are not provided.") .build()); } if (confFileProvided) { final String configFiles = validationContext.getProperty(HADOOP_CONF_FILES).getValue(); ValidationResources resources = validationResourceHolder.get(); if (resources == null || !configFiles.equals(resources.getConfigResources())) { getLogger().debug("Reloading validation resources"); resources = new ValidationResources(configFiles, getConfigurationFromFiles(configFiles)); validationResourceHolder.set(resources); } final Configuration hbaseConfig = resources.getConfiguration(); final String principal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); problems.addAll(KerberosProperties.validatePrincipalAndKeytab( this.getClass().getSimpleName(), hbaseConfig, principal, keytab, getLogger())); } return problems; } }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { boolean confFileProvided = validationContext.getProperty(HADOOP_CONF_FILES).isSet(); boolean zkQuorumProvided = validationContext.getProperty(ZOOKEEPER_QUORUM).isSet(); boolean zkPortProvided = validationContext.getProperty(ZOOKEEPER_CLIENT_PORT).isSet(); boolean znodeParentProvided = validationContext.getProperty(ZOOKEEPER_ZNODE_PARENT).isSet(); boolean retriesProvided = validationContext.getProperty(HBASE_CLIENT_RETRIES).isSet(); final List<ValidationResult> problems = new ArrayList<>(); if (!confFileProvided && (!zkQuorumProvided || !zkPortProvided || !znodeParentProvided || !retriesProvided)) { problems.add(new ValidationResult.Builder() .valid(false) .subject(this.getClass().getSimpleName()) .explanation("ZooKeeper Quorum, ZooKeeper Client Port, ZooKeeper ZNode Parent, and HBase Client Retries are required " + "when Hadoop Configuration Files are not provided.") .build()); } if (confFileProvided) { final String configFiles = validationContext.getProperty(HADOOP_CONF_FILES).getValue(); ValidationResources resources = validationResourceHolder.get(); if (resources == null || !configFiles.equals(resources.getConfigResources())) { getLogger().debug("Reloading validation resources"); resources = new ValidationResources(configFiles, getConfigurationFromFiles(configFiles)); validationResourceHolder.set(resources); } final Configuration hbaseConfig = resources.getConfiguration(); final String principal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); problems.addAll(KerberosProperties.validatePrincipalAndKeytab( this.getClass().getSimpleName(), hbaseConfig, principal, keytab, getLogger())); } return problems; }  }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { boolean confFileProvided = validationContext.getProperty(HADOOP_CONF_FILES).isSet(); boolean zkQuorumProvided = validationContext.getProperty(ZOOKEEPER_QUORUM).isSet(); boolean zkPortProvided = validationContext.getProperty(ZOOKEEPER_CLIENT_PORT).isSet(); boolean znodeParentProvided = validationContext.getProperty(ZOOKEEPER_ZNODE_PARENT).isSet(); boolean retriesProvided = validationContext.getProperty(HBASE_CLIENT_RETRIES).isSet(); final List<ValidationResult> problems = new ArrayList<>(); if (!confFileProvided && (!zkQuorumProvided || !zkPortProvided || !znodeParentProvided || !retriesProvided)) { problems.add(new ValidationResult.Builder() .valid(false) .subject(this.getClass().getSimpleName()) .explanation("ZooKeeper Quorum, ZooKeeper Client Port, ZooKeeper ZNode Parent, and HBase Client Retries are required " + "when Hadoop Configuration Files are not provided.") .build()); } if (confFileProvided) { final String configFiles = validationContext.getProperty(HADOOP_CONF_FILES).getValue(); ValidationResources resources = validationResourceHolder.get(); if (resources == null || !configFiles.equals(resources.getConfigResources())) { getLogger().debug("Reloading validation resources"); resources = new ValidationResources(configFiles, getConfigurationFromFiles(configFiles)); validationResourceHolder.set(resources); } final Configuration hbaseConfig = resources.getConfiguration(); final String principal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); problems.addAll(KerberosProperties.validatePrincipalAndKeytab( this.getClass().getSimpleName(), hbaseConfig, principal, keytab, getLogger())); } return problems; }  @OnEnabled void onEnabled(final ConfigurationContext context); @OnDisabled void shutdown(); @Override void put(final String tableName, final Collection<PutFlowFile> puts); @Override void put(final String tableName, final byte[] rowId, final Collection<PutColumn> columns); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s); }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { boolean confFileProvided = validationContext.getProperty(HADOOP_CONF_FILES).isSet(); boolean zkQuorumProvided = validationContext.getProperty(ZOOKEEPER_QUORUM).isSet(); boolean zkPortProvided = validationContext.getProperty(ZOOKEEPER_CLIENT_PORT).isSet(); boolean znodeParentProvided = validationContext.getProperty(ZOOKEEPER_ZNODE_PARENT).isSet(); boolean retriesProvided = validationContext.getProperty(HBASE_CLIENT_RETRIES).isSet(); final List<ValidationResult> problems = new ArrayList<>(); if (!confFileProvided && (!zkQuorumProvided || !zkPortProvided || !znodeParentProvided || !retriesProvided)) { problems.add(new ValidationResult.Builder() .valid(false) .subject(this.getClass().getSimpleName()) .explanation("ZooKeeper Quorum, ZooKeeper Client Port, ZooKeeper ZNode Parent, and HBase Client Retries are required " + "when Hadoop Configuration Files are not provided.") .build()); } if (confFileProvided) { final String configFiles = validationContext.getProperty(HADOOP_CONF_FILES).getValue(); ValidationResources resources = validationResourceHolder.get(); if (resources == null || !configFiles.equals(resources.getConfigResources())) { getLogger().debug("Reloading validation resources"); resources = new ValidationResources(configFiles, getConfigurationFromFiles(configFiles)); validationResourceHolder.set(resources); } final Configuration hbaseConfig = resources.getConfiguration(); final String principal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); problems.addAll(KerberosProperties.validatePrincipalAndKeytab( this.getClass().getSimpleName(), hbaseConfig, principal, keytab, getLogger())); } return problems; }  @OnEnabled void onEnabled(final ConfigurationContext context); @OnDisabled void shutdown(); @Override void put(final String tableName, final Collection<PutFlowFile> puts); @Override void put(final String tableName, final byte[] rowId, final Collection<PutColumn> columns); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s);  }