@Test public void onTrigger() throws Exception { runner.setProperty(PutHiveStreaming.METASTORE_URI, "thrift: runner.setProperty(PutHiveStreaming.DB_NAME, "default"); runner.setProperty(PutHiveStreaming.TABLE_NAME, "users"); runner.setProperty(PutHiveStreaming.TXNS_PER_BATCH, "100"); Map<String, Object> user1 = new HashMap<String, Object>() { { put("name", "Joe"); put("favorite_number", 146); } }; runner.enqueue(createAvroRecord(Collections.singletonList(user1))); runner.run(); runner.assertTransferCount(PutHiveStreaming.REL_SUCCESS, 1); final MockFlowFile flowFile = runner.getFlowFilesForRelationship(PutHiveStreaming.REL_SUCCESS).get(0); assertEquals("1", flowFile.getAttribute(HIVE_STREAMING_RECORD_COUNT_ATTR)); assertEquals("default.users", flowFile.getAttribute(ATTR_OUTPUT_TABLES)); }
@Override public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { final FunctionContext functionContext = new FunctionContext(context.getProperty(ROLLBACK_ON_FAILURE).asBoolean(), getLogger()); RollbackOnFailure.onTrigger(context, sessionFactory, functionContext, getLogger(), session -> onTrigger(context, session, functionContext)); }
PutHiveStreaming extends AbstractSessionFactoryProcessor { @Override public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { final FunctionContext functionContext = new FunctionContext(context.getProperty(ROLLBACK_ON_FAILURE).asBoolean(), getLogger()); RollbackOnFailure.onTrigger(context, sessionFactory, functionContext, getLogger(), session -> onTrigger(context, session, functionContext)); } }
PutHiveStreaming extends AbstractSessionFactoryProcessor { @Override public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { final FunctionContext functionContext = new FunctionContext(context.getProperty(ROLLBACK_ON_FAILURE).asBoolean(), getLogger()); RollbackOnFailure.onTrigger(context, sessionFactory, functionContext, getLogger(), session -> onTrigger(context, session, functionContext)); }  }
PutHiveStreaming extends AbstractSessionFactoryProcessor { @Override public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { final FunctionContext functionContext = new FunctionContext(context.getProperty(ROLLBACK_ON_FAILURE).asBoolean(), getLogger()); RollbackOnFailure.onTrigger(context, sessionFactory, functionContext, getLogger(), session -> onTrigger(context, session, functionContext)); }  @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); @Override void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory); @OnStopped void cleanup(); }
PutHiveStreaming extends AbstractSessionFactoryProcessor { @Override public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { final FunctionContext functionContext = new FunctionContext(context.getProperty(ROLLBACK_ON_FAILURE).asBoolean(), getLogger()); RollbackOnFailure.onTrigger(context, sessionFactory, functionContext, getLogger(), session -> onTrigger(context, session, functionContext)); }  @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); @Override void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory); @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 PropertyDescriptor CALL_TIMEOUT; static final PropertyDescriptor ROLLBACK_ON_FAILURE; static final Relationship REL_SUCCESS; static final Relationship REL_FAILURE; static final Relationship REL_RETRY; }