@Test public void testFailedPersistState() throws Exception { reset(storage); final ListGCSBucket processor = getProcessor(); final TestRunner runner = buildNewRunner(processor); addRequiredPropertiesToRunner(runner); runner.assertValid(); runner.getStateManager().setFailOnStateSet(Scope.CLUSTER, true); processor.currentKeys = ImmutableSet.of( "test-key-0", "test-key-1" ); processor.currentTimestamp = 4L; assertTrue(runner.getLogger().getErrorMessages().isEmpty()); processor.persistState(runner.getProcessContext()); final List<LogMessage> logMessages = runner.getLogger().getErrorMessages(); assertFalse(logMessages.isEmpty()); assertEquals( 1, logMessages.size() ); }
void persistState(final ProcessContext context) { Map<String, String> state = new HashMap<>(); state.put(CURRENT_TIMESTAMP, String.valueOf(currentTimestamp)); int i = 0; for (String key : currentKeys) { state.put(CURRENT_KEY_PREFIX+i, key); i++; } try { context.getStateManager().setState(state, Scope.CLUSTER); } catch (IOException ioe) { getLogger().error("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", ioe); } }
ListGCSBucket extends AbstractGCSProcessor { void persistState(final ProcessContext context) { Map<String, String> state = new HashMap<>(); state.put(CURRENT_TIMESTAMP, String.valueOf(currentTimestamp)); int i = 0; for (String key : currentKeys) { state.put(CURRENT_KEY_PREFIX+i, key); i++; } try { context.getStateManager().setState(state, Scope.CLUSTER); } catch (IOException ioe) { getLogger().error("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", ioe); } } }
ListGCSBucket extends AbstractGCSProcessor { void persistState(final ProcessContext context) { Map<String, String> state = new HashMap<>(); state.put(CURRENT_TIMESTAMP, String.valueOf(currentTimestamp)); int i = 0; for (String key : currentKeys) { state.put(CURRENT_KEY_PREFIX+i, key); i++; } try { context.getStateManager().setState(state, Scope.CLUSTER); } catch (IOException ioe) { getLogger().error("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", ioe); } }  }
ListGCSBucket extends AbstractGCSProcessor { void persistState(final ProcessContext context) { Map<String, String> state = new HashMap<>(); state.put(CURRENT_TIMESTAMP, String.valueOf(currentTimestamp)); int i = 0; for (String key : currentKeys) { state.put(CURRENT_KEY_PREFIX+i, key); i++; } try { context.getStateManager().setState(state, Scope.CLUSTER); } catch (IOException ioe) { getLogger().error("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", ioe); } }  @Override List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override Set<Relationship> getRelationships(); @Override void onTrigger(ProcessContext context, ProcessSession session); }
ListGCSBucket extends AbstractGCSProcessor { void persistState(final ProcessContext context) { Map<String, String> state = new HashMap<>(); state.put(CURRENT_TIMESTAMP, String.valueOf(currentTimestamp)); int i = 0; for (String key : currentKeys) { state.put(CURRENT_KEY_PREFIX+i, key); i++; } try { context.getStateManager().setState(state, Scope.CLUSTER); } catch (IOException ioe) { getLogger().error("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", ioe); } }  @Override List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override Set<Relationship> getRelationships(); @Override void onTrigger(ProcessContext context, ProcessSession session); static final PropertyDescriptor BUCKET; static final PropertyDescriptor PREFIX; static final PropertyDescriptor USE_GENERATIONS; static final Set<Relationship> relationships; static final String CURRENT_TIMESTAMP; static final String CURRENT_KEY_PREFIX; }