@Test public void testRestorePreviousState() throws Exception { reset(storage); final ListGCSBucket processor = getProcessor(); final TestRunner runner = buildNewRunner(processor); addRequiredPropertiesToRunner(runner); runner.assertValid(); final Map<String, String> state = ImmutableMap.of( ListGCSBucket.CURRENT_TIMESTAMP, String.valueOf(4L), ListGCSBucket.CURRENT_KEY_PREFIX + "0", "test-key-0", ListGCSBucket.CURRENT_KEY_PREFIX + "1", "test-key-1" ); runner.getStateManager().setState(state, Scope.CLUSTER); assertNull(processor.currentKeys); assertEquals( 0L, processor.currentTimestamp ); processor.restoreState(runner.getProcessContext()); assertNotNull(processor.currentKeys); assertTrue(processor.currentKeys.contains("test-key-0")); assertTrue(processor.currentKeys.contains("test-key-1")); assertEquals( 4L, processor.currentTimestamp ); }
void restoreState(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() == -1L || stateMap.get(CURRENT_TIMESTAMP) == null || stateMap.get(CURRENT_KEY_PREFIX+"0") == null) { currentTimestamp = 0L; currentKeys = new HashSet<>(); } else { currentTimestamp = Long.parseLong(stateMap.get(CURRENT_TIMESTAMP)); currentKeys = extractKeys(stateMap); } }
ListGCSBucket extends AbstractGCSProcessor { void restoreState(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() == -1L || stateMap.get(CURRENT_TIMESTAMP) == null || stateMap.get(CURRENT_KEY_PREFIX+"0") == null) { currentTimestamp = 0L; currentKeys = new HashSet<>(); } else { currentTimestamp = Long.parseLong(stateMap.get(CURRENT_TIMESTAMP)); currentKeys = extractKeys(stateMap); } } }
ListGCSBucket extends AbstractGCSProcessor { void restoreState(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() == -1L || stateMap.get(CURRENT_TIMESTAMP) == null || stateMap.get(CURRENT_KEY_PREFIX+"0") == null) { currentTimestamp = 0L; currentKeys = new HashSet<>(); } else { currentTimestamp = Long.parseLong(stateMap.get(CURRENT_TIMESTAMP)); currentKeys = extractKeys(stateMap); } }  }
ListGCSBucket extends AbstractGCSProcessor { void restoreState(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() == -1L || stateMap.get(CURRENT_TIMESTAMP) == null || stateMap.get(CURRENT_KEY_PREFIX+"0") == null) { currentTimestamp = 0L; currentKeys = new HashSet<>(); } else { currentTimestamp = Long.parseLong(stateMap.get(CURRENT_TIMESTAMP)); currentKeys = extractKeys(stateMap); } }  @Override List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override Set<Relationship> getRelationships(); @Override void onTrigger(ProcessContext context, ProcessSession session); }
ListGCSBucket extends AbstractGCSProcessor { void restoreState(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() == -1L || stateMap.get(CURRENT_TIMESTAMP) == null || stateMap.get(CURRENT_KEY_PREFIX+"0") == null) { currentTimestamp = 0L; currentKeys = new HashSet<>(); } else { currentTimestamp = Long.parseLong(stateMap.get(CURRENT_TIMESTAMP)); currentKeys = extractKeys(stateMap); } }  @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; }