@Test public void testBecomePrimaryWithNewerLocalState() throws InitializationException { final long now = System.currentTimeMillis(); final Map<String, String> cells = new HashMap<>(); cells.put("greeting", "hello"); cells.put("name", "nifi"); hBaseClient.addResult("row0", cells, now - 2); hBaseClient.addResult("row1", cells, now - 1); hBaseClient.addResult("row2", cells, now - 1); hBaseClient.addResult("row3", cells, now); runner.run(100); runner.assertAllFlowFilesTransferred(GetHBase.REL_SUCCESS, 4); final MockCacheClient otherCacheClient = new MockCacheClient(); runner.addControllerService("otherCacheClient", otherCacheClient); runner.enableControllerService(otherCacheClient); runner.setProperty(GetHBase.DISTRIBUTED_CACHE_SERVICE, "otherCacheClient"); hBaseClient.addResult("row4", cells, now + 1); runner.run(); runner.assertAllFlowFilesTransferred(GetHBase.REL_SUCCESS, 5); runner.setProperty(GetHBase.DISTRIBUTED_CACHE_SERVICE, "cacheClient"); proc.onPrimaryNodeChange(PrimaryNodeState.ELECTED_PRIMARY_NODE); hBaseClient.addResult("row0", cells, now - 2); hBaseClient.addResult("row1", cells, now - 1); hBaseClient.addResult("row2", cells, now - 1); hBaseClient.addResult("row3", cells, now); hBaseClient.addResult("row4", cells, now + 1); runner.clearTransferState(); runner.run(100); runner.assertAllFlowFilesTransferred(GetHBase.REL_SUCCESS, 0); }
@OnPrimaryNodeStateChange public void onPrimaryNodeChange(final PrimaryNodeState newState) { justElectedPrimaryNode = (newState == PrimaryNodeState.ELECTED_PRIMARY_NODE); }
GetHBase extends AbstractProcessor { @OnPrimaryNodeStateChange public void onPrimaryNodeChange(final PrimaryNodeState newState) { justElectedPrimaryNode = (newState == PrimaryNodeState.ELECTED_PRIMARY_NODE); } }
GetHBase extends AbstractProcessor { @OnPrimaryNodeStateChange public void onPrimaryNodeChange(final PrimaryNodeState newState) { justElectedPrimaryNode = (newState == PrimaryNodeState.ELECTED_PRIMARY_NODE); }  }
GetHBase extends AbstractProcessor { @OnPrimaryNodeStateChange public void onPrimaryNodeChange(final PrimaryNodeState newState) { justElectedPrimaryNode = (newState == PrimaryNodeState.ELECTED_PRIMARY_NODE); }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void parseColumns(final ProcessContext context); @OnPrimaryNodeStateChange void onPrimaryNodeChange(final PrimaryNodeState newState); @OnRemoved void onRemoved(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
GetHBase extends AbstractProcessor { @OnPrimaryNodeStateChange public void onPrimaryNodeChange(final PrimaryNodeState newState) { justElectedPrimaryNode = (newState == PrimaryNodeState.ELECTED_PRIMARY_NODE); }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void parseColumns(final ProcessContext context); @OnPrimaryNodeStateChange void onPrimaryNodeChange(final PrimaryNodeState newState); @OnRemoved void onRemoved(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session);  }