Skip to content

NIFI-9362: Ensure that we update the StateMap in AbstractListProcesso… #5507

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
markap14 wants to merge 1 commit into from
Closed

NIFI-9362: Ensure that we update the StateMap in AbstractListProcesso… #5507

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 14 additions & 11 deletions ...cessor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -851,6 +851,20 @@ public void listByTrackingTimestamps(final ProcessContext context, final Process
if (latestListedEntryTimestampThisCycleMillis != null) {
final boolean processedNewFiles = entitiesListed > 0;

if (processedNewFiles) {
// If there have been files created, update the last timestamp we processed.
// Retrieving lastKey instead of using latestListedEntryTimestampThisCycleMillis is intentional here,
// because latestListedEntryTimestampThisCycleMillis might be removed if it's not old enough.
if (!orderedEntries.lastKey().equals(lastProcessedLatestEntryTimestampMillis)) {
// If the latest timestamp at this cycle becomes different than the previous one, we need to clear identifiers.
// If it didn't change, we need to add identifiers.
latestIdentifiersProcessed.clear();
}
// Capture latestIdentifierProcessed.
latestIdentifiersProcessed.addAll(orderedEntries.lastEntry().getValue().stream().map(T::getIdentifier).collect(Collectors.toList()));
lastProcessedLatestEntryTimestampMillis = orderedEntries.lastKey();
}

if (!latestListedEntryTimestampThisCycleMillis.equals(lastListedLatestEntryTimestampMillis) || processedNewFiles) {
// We have performed a listing and pushed any FlowFiles out that may have been generated
// Now, we need to persist state about the Last Modified timestamp of the newest file
Expand All @@ -870,17 +884,6 @@ public void listByTrackingTimestamps(final ProcessContext context, final Process
}

if (processedNewFiles) {
// If there have been files created, update the last timestamp we processed.
// Retrieving lastKey instead of using latestListedEntryTimestampThisCycleMillis is intentional here,
// because latestListedEntryTimestampThisCycleMillis might be removed if it's not old enough.
if (!orderedEntries.lastKey().equals(lastProcessedLatestEntryTimestampMillis)) {
// If the latest timestamp at this cycle becomes different than the previous one, we need to clear identifiers.
// If it didn't change, we need to add identifiers.
latestIdentifiersProcessed.clear();
}
// Capture latestIdentifierProcessed.
latestIdentifiersProcessed.addAll(orderedEntries.lastEntry().getValue().stream().map(T::getIdentifier).collect(Collectors.toList()));
lastProcessedLatestEntryTimestampMillis = orderedEntries.lastKey();
getLogger().info("Successfully created listing with {} new objects", new Object[]{entitiesListed});
session.commitAsync();
}
Expand Down
41 changes: 41 additions & 0 deletions ...or-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.nifi.components.Validator;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.distributed.cache.client.Deserializer;
Expand Down Expand Up @@ -115,6 +116,46 @@ public void setup() {
@Rule
public final TemporaryFolder testFolder = new TemporaryFolder();

@Test
public void testStateMigratedWhenPrimaryNodeSwitch() throws IOException {
// add a few entities
for (int i=0; i < 5; i++) {
proc.addEntity(String.valueOf(i), String.valueOf(i), 88888L);
}

// Add an entity with a later timestamp
proc.addEntity("10", "10", 99999999L);

// Run the processor. All 6 should be listed.
runner.run();
runner.assertAllFlowFilesTransferred(AbstractListProcessor.REL_SUCCESS, 6);

// Now, we want to mimic Primary Node changing. To do so, we'll capture the Cluster State from the State Manager,
// create a new Processor, and set the state to be the same, and update the processor in order to produce the same listing.
final ConcreteListProcessor secondProc = new ConcreteListProcessor();
// Add same listing to the new processor
for (int i=0; i < 5; i++) {
secondProc.addEntity(String.valueOf(i), String.valueOf(i), 88888L);
}
secondProc.addEntity("10", "10", 99999999L);

// Create new runner for the second processor and update its state to match that of the last TestRunner.
final StateMap stateMap = runner.getStateManager().getState(Scope.CLUSTER);
runner = TestRunners.newTestRunner(secondProc);
runner.getStateManager().setState(stateMap.toMap(), Scope.CLUSTER);

// Run several times, ensuring that nothing is emitted.
for (int i=0; i < 10; i++) {
runner.run();
runner.assertAllFlowFilesTransferred(AbstractListProcessor.REL_SUCCESS, 0);
}

// Add one more entry and ensure that it's emitted.
secondProc.addEntity("new", "new", 999999990L);
runner.run();
runner.assertAllFlowFilesTransferred(AbstractListProcessor.REL_SUCCESS, 1);
}

@Test
public void testStateMigratedFromCacheService() throws InitializationException {

Expand Down