### Question:
PutHiveStreaming extends AbstractProcessor { @OnScheduled public void setup(final ProcessContext context) { ComponentLog log = getLogger(); final String metastoreUri = context.getProperty(METASTORE_URI).getValue(); final String dbName = context.getProperty(DB_NAME).getValue(); final String tableName = context.getProperty(TABLE_NAME).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final Integer maxConnections = context.getProperty(MAX_OPEN_CONNECTIONS).asInteger(); final Integer heartbeatInterval = context.getProperty(HEARTBEAT_INTERVAL).asInteger(); final Integer txnsPerBatch = context.getProperty(TXNS_PER_BATCH).asInteger(); final String configFiles = context.getProperty(HIVE_CONFIGURATION_RESOURCES).getValue(); hiveConfig = hiveConfigurator.getConfigurationFromFiles(configFiles); for (final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) { final PropertyDescriptor descriptor = entry.getKey(); if (descriptor.isDynamic()) { hiveConfig.set(descriptor.getName(), entry.getValue()); } } options = new HiveOptions(metastoreUri, dbName, tableName) .withTxnsPerBatch(txnsPerBatch) .withAutoCreatePartitions(autoCreatePartitions) .withMaxOpenConnections(maxConnections) .withHeartBeatInterval(heartbeatInterval); hiveConfigurator.preload(hiveConfig); if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final String principal = context.getProperty(kerberosProperties.getKerberosPrincipal()).getValue(); final String keyTab = context.getProperty(kerberosProperties.getKerberosKeytab()).getValue(); log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[]{principal, keyTab}); try { ugi = hiveConfigurator.authenticate(hiveConfig, principal, keyTab, TICKET_RENEWAL_PERIOD, log); } catch (AuthenticationFailedException ae) { throw new ProcessException("Kerberos authentication failed for Hive Streaming", ae); } log.info("Successfully logged in as principal {} with keytab {}", new Object[]{principal, keyTab}); options = options.withKerberosPrincipal(principal).withKerberosKeytab(keyTab); } else { ugi = null; } allWriters = new ConcurrentHashMap<>(); String timeoutName = "put-hive-streaming-%d"; this.callTimeoutPool = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder().setNameFormat(timeoutName).build()); sendHeartBeat.set(true); heartBeatTimer = new Timer(); setupHeartBeatTimer(); }  @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); @Override void onTrigger(ProcessContext context, ProcessSession session); @OnStopped void cleanup(); static final String HIVE_STREAMING_RECORD_COUNT_ATTR; static final PropertyDescriptor METASTORE_URI; static final PropertyDescriptor HIVE_CONFIGURATION_RESOURCES; static final PropertyDescriptor DB_NAME; static final PropertyDescriptor TABLE_NAME; static final PropertyDescriptor PARTITION_COLUMNS; static final PropertyDescriptor AUTOCREATE_PARTITIONS; static final PropertyDescriptor MAX_OPEN_CONNECTIONS; static final PropertyDescriptor HEARTBEAT_INTERVAL; static final PropertyDescriptor TXNS_PER_BATCH; static final PropertyDescriptor RECORDS_PER_TXN; static final Relationship REL_SUCCESS; static final Relationship REL_FAILURE; static final Relationship REL_RETRY; }### Answer:
@Test public void testSetup() throws Exception { runner.setValidateExpressionUsage(false); runner.assertNotValid(); runner.setProperty(PutHiveStreaming.METASTORE_URI, "thrift: runner.setProperty(PutHiveStreaming.DB_NAME, "default"); runner.assertNotValid(); runner.setProperty(PutHiveStreaming.TABLE_NAME, "users"); runner.assertValid(); runner.run(); }