Fix PeerForwardingProcessorDecorator to process records locally when exclude identification keys is set #5178
Conversation
…exclude identification keys is set Signed-off-by: Krishna Kondaka <[email protected]>
kkondaka
requested review from
sb2k16,
chenqi0805,
engechas,
graytaylor0,
dinujoh,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
| @@ -101,7 +101,12 @@ public Collection<Record<Event>> execute(final Collection<Record<Event>> records | |||
| final Collection<Record<Event>> recordsSkipped = new ArrayList<>(); | |||
| for (Record<Event> record: records) { | |||
| if (((RequiresPeerForwarding)innerProcessor).isApplicableEventForPeerForwarding(record.getData())) { | |||
| recordsToProcess.add(record); | |||
| if (isPeerForwardingDisabled()) { | |||
| System.out.println("=====PROCESSING LOCALLY===="); | |||
There was a problem hiding this comment.
You have a println here. I expect the build to fail as a result.
| } | ||
| verify(peerForwarderProvider, times(1)).register(pipelineName, processor, pluginId, identificationKeys, PIPELINE_WORKER_THREADS); | ||
| verifyNoMoreInteractions(peerForwarderProvider); | ||
| Collection<Record<Event>> result = processors.get(0).execute(testData); |
There was a problem hiding this comment.
I'm not following this test. It seems that you mock this above on line 190 and then assert it here.
I'm not 100% confident that this test is covering the situation.
There was a problem hiding this comment.
OK. I think I see now. The processors are the decorated processors under test.
Maybe rename to objectsUnderTest.
| } | ||
| verify(peerForwarderProvider, times(1)).register(pipelineName, processor, pluginId, identificationKeys, PIPELINE_WORKER_THREADS); | ||
| verifyNoMoreInteractions(peerForwarderProvider); | ||
| Collection<Record<Event>> result = processors.get(0).execute(testData); |
There was a problem hiding this comment.
OK. I think I see now. The processors are the decorated processors under test.
Maybe rename to objectsUnderTest.
| result = processors.get(1).execute(testData); | ||
| assertThat(result.size(), equalTo(testData.size())); | ||
| assertThat(result, equalTo(testData)); | ||
| } |
There was a problem hiding this comment.
Should we also have some form of verify(processor1).execute? and verify(processor2).execute?
There was a problem hiding this comment.
Why do verify when I am doing processors.get(1).execute(testData);?
There was a problem hiding this comment.
The reason is that processor1 != processors.get(0). The value processor1 is the inner processor. And processors.get(0) is the actual processor. So you can't say for sure in the current tests that the data reached processor1.
| when(requiresPeerForwarding.isApplicableEventForPeerForwarding(event)).thenReturn(true); | ||
| when(requiresPeerForwardingCopy.isApplicableEventForPeerForwarding(event)).thenReturn(true); | ||
|
|
||
| Processor processor1 = (Processor)requiresPeerForwarding; |
There was a problem hiding this comment.
Maybe rename these to innerProcessor1 and innerProcessor2 to provide clarity.
Signed-off-by: Krishna Kondaka <[email protected]>
| @@ -101,7 +101,11 @@ public Collection<Record<Event>> execute(final Collection<Record<Event>> records | |||
| final Collection<Record<Event>> recordsSkipped = new ArrayList<>(); | |||
| for (Record<Event> record: records) { | |||
| if (((RequiresPeerForwarding)innerProcessor).isApplicableEventForPeerForwarding(record.getData())) { | |||
| recordsToProcess.add(record); | |||
| if (isPeerForwardingDisabled()) { | |||
There was a problem hiding this comment.
Nit: This condition could just be a part of the above if statement
if (!isPeerForwardingDisabled && ((RequiresPeerForwarding)innerProcessor).isApplicableEventForPeerForwarding(record.getData()))
There was a problem hiding this comment.
I thought about it but the next condition has isForLocalProcessingOnly(record.getData())){ which will fail. This is more clean
Description
Fix PeerForwardingProcessorDecorator to process records locally when exclude identification keys is set.
The changes in PR 5127 missed a step in execute() which is supposed to process records locally if peer forwarding is disabled for a processor
Issues Resolved
Resolves #[Issue number to be closed when this PR is merged]
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.