Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-9437 Flowfile Expiration cannot exceed 24 days 

NIFI-9437 Addressing review comments, correcting setter in StatelessFlowQueue class

Signed-off-by: Nathan Gough <thenatog@gmail.com>

This closes #6558.
This commit is contained in:
Timea Barna 2022-10-20 12:06:31 +02:00 committed by Nathan Gough
parent 753cb1b9df
commit 7c33516aee
9 changed files with 13 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java
View File

@ -177,7 +177,7 @@ public interface FlowFileQueue {
String getFlowFileExpiration(); String getFlowFileExpiration();
int getFlowFileExpiration(TimeUnit timeUnit); long getFlowFileExpiration(TimeUnit timeUnit);
void setFlowFileExpiration(String flowExpirationPeriod); void setFlowFileExpiration(String flowExpirationPeriod);

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/AbstractFlowFileQueue.java
View File

@ -98,8 +98,8 @@ public abstract class AbstractFlowFileQueue implements FlowFileQueue {
} }
@Override @Override
public int getFlowFileExpiration(final TimeUnit timeUnit) { public long getFlowFileExpiration(final TimeUnit timeUnit) {
return (int) timeUnit.convert(expirationPeriod.get().getMillis(), TimeUnit.MILLISECONDS); return timeUnit.convert(expirationPeriod.get().getMillis(), TimeUnit.MILLISECONDS);
} }
@Override @Override

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/partition/SwappablePriorityQueueLocalPartition.java
View File

@ -138,7 +138,7 @@ public class SwappablePriorityQueueLocalPartition implements LocalQueuePartition
return poll(filter, expiredRecords, PollStrategy.UNPENALIZED_FLOWFILES); return poll(filter, expiredRecords, PollStrategy.UNPENALIZED_FLOWFILES);
} }
private int getExpiration() { private long getExpiration() {
return flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS); return flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS);
} }

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
View File

@ -262,7 +262,7 @@ public class TestWriteAheadFlowFileRepository {
} }
@Override @Override
public int getFlowFileExpiration(TimeUnit timeUnit) { public long getFlowFileExpiration(TimeUnit timeUnit) {
return 0; return 0;
} }

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/search/attributematchers/ExpirationMatcher.java
View File

@ -35,7 +35,7 @@ public class ExpirationMatcher implements AttributeMatcher<Connection> {
@Override @Override
public void match(final Connection component, final SearchQuery query, final List<String> matches) { public void match(final Connection component, final SearchQuery query, final List<String> matches) {
if (containsKeyword(query)) { if (containsKeyword(query)) {
final int expirationMillis = component.getFlowFileQueue().getFlowFileExpiration(TimeUnit.MILLISECONDS); final long expirationMillis = component.getFlowFileQueue().getFlowFileExpiration(TimeUnit.MILLISECONDS);
if (expirationMillis > 0) { if (expirationMillis > 0) {
matches.add(MATCH_PREFIX + component.getFlowFileQueue().getFlowFileExpiration()); matches.add(MATCH_PREFIX + component.getFlowFileQueue().getFlowFileExpiration());

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/controller/ComponentMockUtil.java
View File

@ -226,7 +226,7 @@ public class ComponentMockUtil {
final Optional<String> versionedId, final Optional<String> versionedId,
final Collection<Relationship> relationships, final Collection<Relationship> relationships,
final List<FlowFilePrioritizer> flowFilePrioritizers, final List<FlowFilePrioritizer> flowFilePrioritizers,
final int flowFileExpirationInMs, final long flowFileExpirationInMs,
final String backPressureDataSize, final String backPressureDataSize,
final long backPressureCount, final long backPressureCount,
final Connectable source, final Connectable source,

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/controller/ControllerSearchServiceIntegrationTest.java
View File

@ -509,7 +509,7 @@ public class ControllerSearchServiceIntegrationTest extends AbstractControllerSe
final Connection connection = getConnection("connection", "connectionName", getBasicRelationships(), processor1, processor2, AUTHORIZED); final Connection connection = getConnection("connection", "connectionName", getBasicRelationships(), processor1, processor2, AUTHORIZED);
final FlowFileQueue flowFileQueue = Mockito.mock(FlowFileQueue.class); final FlowFileQueue flowFileQueue = Mockito.mock(FlowFileQueue.class);
Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(5); Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(5L);
Mockito.when(flowFileQueue.getFlowFileExpiration()).thenReturn("5"); Mockito.when(flowFileQueue.getFlowFileExpiration()).thenReturn("5");
Mockito.when(connection.getFlowFileQueue()).thenReturn(flowFileQueue); Mockito.when(connection.getFlowFileQueue()).thenReturn(flowFileQueue);

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/search/attributematchers/ExpirationMatcherTest.java
View File

@ -96,11 +96,11 @@ public class ExpirationMatcherTest extends AbstractAttributeMatcherTest{
} }
private void givenExpired() { private void givenExpired() {
Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(5); Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(5L);
Mockito.when(flowFileQueue.getFlowFileExpiration()).thenReturn("5"); Mockito.when(flowFileQueue.getFlowFileExpiration()).thenReturn("5");
} }
private void givenNotExpired() { private void givenNotExpired() {
Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(0); Mockito.when(flowFileQueue.getFlowFileExpiration(TimeUnit.MILLISECONDS)).thenReturn(0L);
} }
} }

6
nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/queue/StatelessFlowFileQueue.java
View File

@ -283,13 +283,13 @@ public class StatelessFlowFileQueue implements DrainableFlowFileQueue {
} }
@Override @Override
public int getFlowFileExpiration(final TimeUnit timeUnit) { public long getFlowFileExpiration(final TimeUnit timeUnit) {
return (int) timeUnit.convert(expirationMillis, TimeUnit.MILLISECONDS); return timeUnit.convert(expirationMillis, TimeUnit.MILLISECONDS);
} }
@Override @Override
public void setFlowFileExpiration(final String flowExpirationPeriod) { public void setFlowFileExpiration(final String flowExpirationPeriod) {
expirationMillis = (int) FormatUtils.getPreciseTimeDuration(flowExpirationPeriod, TimeUnit.MILLISECONDS); expirationMillis = Math.round(FormatUtils.getPreciseTimeDuration(flowExpirationPeriod, TimeUnit.MILLISECONDS));
} }
@Override @Override