http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java index 658e8c0..8d398aa 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java @@ -80,6 +80,7 @@ import org.apache.nifi.provenance.ProvenanceEventRecord; import org.apache.nifi.provenance.ProvenanceEventRepository; import org.apache.nifi.provenance.ProvenanceEventType; import org.apache.nifi.stream.io.StreamUtils; +import org.apache.nifi.util.NiFiProperties; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -138,7 +139,7 @@ public class TestStandardProcessSession { public void setup() throws IOException { resourceClaimManager = new StandardResourceClaimManager(); - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestStandardProcessSession.class.getResource("/conf/nifi.properties").getFile()); final FlowFileEventRepository flowFileEventRepo = Mockito.mock(FlowFileEventRepository.class); final CounterRepository counterRepo = Mockito.mock(CounterRepository.class); provenanceRepo = new MockProvenanceRepository(); @@ -192,9 +193,9 @@ public class TestStandardProcessSession { final Relationship relationship = (Relationship) arguments[0]; if (relationship == Relationship.SELF) { return Collections.emptySet(); - } else if (relationship == FAKE_RELATIONSHIP || relationship.equals(FAKE_RELATIONSHIP) ){ + } else if (relationship == FAKE_RELATIONSHIP || relationship.equals(FAKE_RELATIONSHIP)) { return null; - }else { + } else { return new HashSet<>(connList); } } @@ -213,10 +214,10 @@ public class TestStandardProcessSession { @Test public void testAppendToChildThrowsIOExceptionThenRemove() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .id(1000L) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .id(1000L) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile original = session.get(); assertNotNull(original); @@ -245,10 +246,10 @@ public class TestStandardProcessSession { @Test public void testWriteForChildThrowsIOExceptionThenRemove() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .id(1000L) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .id(1000L) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile original = session.get(); assertNotNull(original); @@ -274,7 +275,6 @@ public class TestStandardProcessSession { assertEquals(0, numClaims); } - @Test public void testModifyContentThenRollback() throws IOException { assertEquals(0, contentRepo.getExistingClaims().size()); @@ -283,10 +283,10 @@ public class TestStandardProcessSession { assertEquals(1, contentRepo.getExistingClaims().size()); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); @@ -384,10 +384,10 @@ public class TestStandardProcessSession { public void testAppendAfterSessionClosesStream() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); assertNotNull(flowFile); @@ -405,10 +405,10 @@ public class TestStandardProcessSession { public void testExportTo() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); assertNotNull(flowFile); @@ -440,15 +440,15 @@ public class TestStandardProcessSession { public void testReadAfterSessionClosesStream() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); final FlowFile flowFile = session.get(); assertNotNull(flowFile); final AtomicReference<InputStream> inputStreamHolder = new AtomicReference<>(null); - session.read(flowFile, true , new InputStreamCallback() { + session.read(flowFile, true, new InputStreamCallback() { @Override public void process(final InputStream inputStream) throws IOException { inputStreamHolder.set(inputStream); @@ -461,10 +461,10 @@ public class TestStandardProcessSession { public void testStreamAfterSessionClosesStream() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); assertNotNull(flowFile); @@ -485,10 +485,10 @@ public class TestStandardProcessSession { public void testWriteAfterSessionClosesStream() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); assertNotNull(flowFile); @@ -506,9 +506,9 @@ public class TestStandardProcessSession { public void testCreateThenRollbackRemovesContent() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); final StreamCallback nop = new StreamCallback() { @@ -536,9 +536,9 @@ public class TestStandardProcessSession { @Test public void testForksNotEmittedIfFilesDeleted() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); @@ -553,9 +553,9 @@ public class TestStandardProcessSession { @Test public void testProvenanceEventsEmittedForForkIfNotRemoved() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); @@ -570,9 +570,9 @@ public class TestStandardProcessSession { @Test public void testProvenanceEventsEmittedForRemove() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); @@ -590,10 +590,10 @@ public class TestStandardProcessSession { public void testUuidAttributeCannotBeUpdated() { String originalUuid = "11111111-1111-1111-1111-111111111111"; final FlowFileRecord flowFileRecord1 = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", originalUuid) - .entryDate(System.currentTimeMillis()) - .build(); + .id(1L) + .addAttribute("uuid", originalUuid) + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord1); @@ -625,16 +625,16 @@ public class TestStandardProcessSession { @Test public void testUpdateAttributesThenJoin() throws IOException { final FlowFileRecord flowFileRecord1 = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") - .entryDate(System.currentTimeMillis()) - .build(); + .id(1L) + .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") + .entryDate(System.currentTimeMillis()) + .build(); final FlowFileRecord flowFileRecord2 = new StandardFlowFileRecord.Builder() - .id(2L) - .addAttribute("uuid", "22222222-2222-2222-2222-222222222222") - .entryDate(System.currentTimeMillis()) - .build(); + .id(2L) + .addAttribute("uuid", "22222222-2222-2222-2222-222222222222") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord1); flowFileQueue.put(flowFileRecord2); @@ -698,9 +698,9 @@ public class TestStandardProcessSession { @Test public void testForkOneToOneReported() throws IOException { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord); @@ -815,11 +815,11 @@ public class TestStandardProcessSession { @Test public void testMissingFlowFileExceptionThrownWhenUnableToReadData() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - .size(1L) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .size(1L) + .build(); flowFileQueue.put(flowFileRecord); // attempt to read the data. @@ -894,17 +894,16 @@ public class TestStandardProcessSession { assertEquals(1, countAfterAppend); } - @Test @SuppressWarnings("unchecked") public void testExpireDecrementsClaimsOnce() throws IOException { final ContentClaim contentClaim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(contentClaim) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(contentClaim) + .build(); Mockito.doAnswer(new Answer<List<FlowFileRecord>>() { int iterations = 0; @@ -930,7 +929,7 @@ public class TestStandardProcessSession { public void testManyFilesOpened() throws IOException { StandardProcessSession[] standardProcessSessions = new StandardProcessSession[100000]; - for(int i = 0; i<70000;i++){ + for (int i = 0; i < 70000; i++) { standardProcessSessions[i] = new StandardProcessSession(context); FlowFile flowFile = standardProcessSessions[i].create(); @@ -950,12 +949,12 @@ public class TestStandardProcessSession { StreamUtils.fillBuffer(in, buff); } }); - } catch (Exception e){ - System.out.println("Failed at file:"+i); + } catch (Exception e) { + System.out.println("Failed at file:" + i); throw e; } - if(i%1000==0){ - System.out.println("i:"+i); + if (i % 1000 == 0) { + System.out.println("i:" + i); } } } @@ -963,11 +962,11 @@ public class TestStandardProcessSession { @Test public void testMissingFlowFileExceptionThrownWhenUnableToReadDataStreamCallback() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - .size(1L) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .size(1L) + .build(); flowFileQueue.put(flowFileRecord); // attempt to read the data. @@ -987,10 +986,10 @@ public class TestStandardProcessSession { @Test public void testContentNotFoundExceptionThrownWhenUnableToReadDataStreamCallbackOffsetTooLarge() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .build(); flowFileQueue.put(flowFileRecord); FlowFile ff1 = session.get(); @@ -1003,12 +1002,12 @@ public class TestStandardProcessSession { session.commit(); final FlowFileRecord flowFileRecord2 = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - .contentClaimOffset(1000L) - .size(1000L) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .contentClaimOffset(1000L) + .size(1000L) + .build(); flowFileQueue.put(flowFileRecord2); // attempt to read the data. @@ -1028,10 +1027,10 @@ public class TestStandardProcessSession { @Test public void testContentNotFoundExceptionThrownWhenUnableToReadDataOffsetTooLarge() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - .build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .build(); flowFileQueue.put(flowFileRecord); @@ -1045,11 +1044,10 @@ public class TestStandardProcessSession { session.commit(); final FlowFileRecord flowFileRecord2 = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - - .contentClaimOffset(1000L).size(1L).build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .contentClaimOffset(1000L).size(1L).build(); flowFileQueue.put(flowFileRecord2); // attempt to read the data. @@ -1110,15 +1108,13 @@ public class TestStandardProcessSession { } } - @Test public void testCommitFailureRequeuesFlowFiles() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - - .contentClaimOffset(0L).size(0L).build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .contentClaimOffset(0L).size(0L).build(); flowFileQueue.put(flowFileRecord); final FlowFile originalFlowFile = session.get(); @@ -1152,11 +1148,10 @@ public class TestStandardProcessSession { @Test public void testRollbackAfterCheckpoint() { final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) - - .contentClaimOffset(0L).size(0L).build(); + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .contentClaim(new StandardContentClaim(resourceClaimManager.newResourceClaim("x", "x", "0", true), 0L)) + .contentClaimOffset(0L).size(0L).build(); flowFileQueue.put(flowFileRecord); final FlowFile originalFlowFile = session.get(); @@ -1246,9 +1241,9 @@ public class TestStandardProcessSession { @Test public void testContentModifiedEmittedAndNotAttributesModified() throws IOException { final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", "000000000000-0000-0000-0000-00000000") - .build(); + .id(1L) + .addAttribute("uuid", "000000000000-0000-0000-0000-00000000") + .build(); this.flowFileQueue.put(flowFile); FlowFile existingFlowFile = session.get(); @@ -1273,9 +1268,9 @@ public class TestStandardProcessSession { public void testGetWithCount() { for (int i = 0; i < 8; i++) { final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder() - .id(i) - .addAttribute("uuid", "000000000000-0000-0000-0000-0000000" + i) - .build(); + .id(i) + .addAttribute("uuid", "000000000000-0000-0000-0000-0000000" + i) + .build(); this.flowFileQueue.put(flowFile); } @@ -1286,9 +1281,9 @@ public class TestStandardProcessSession { @Test public void testAttributesModifiedEmitted() throws IOException { final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", "000000000000-0000-0000-0000-00000000") - .build(); + .id(1L) + .addAttribute("uuid", "000000000000-0000-0000-0000-00000000") + .build(); this.flowFileQueue.put(flowFile); FlowFile existingFlowFile = session.get(); @@ -1351,11 +1346,11 @@ public class TestStandardProcessSession { } final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() - .contentClaim(claim) - .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") - .entryDate(System.currentTimeMillis()) - .size(12L) - .build(); + .contentClaim(claim) + .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") + .entryDate(System.currentTimeMillis()) + .size(12L) + .build(); flowFileQueue.put(flowFileRecord); final FlowFile flowFile = session.get(); @@ -1377,10 +1372,10 @@ public class TestStandardProcessSession { @Test public void testTransferUnknownRelationship() { final FlowFileRecord flowFileRecord1 = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") - .entryDate(System.currentTimeMillis()) - .build(); + .id(1L) + .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") + .entryDate(System.currentTimeMillis()) + .build(); flowFileQueue.put(flowFileRecord1); @@ -1403,6 +1398,7 @@ public class TestStandardProcessSession { } private static class MockFlowFileRepository implements FlowFileRepository { + private boolean failOnUpdate = false; private final AtomicLong idGenerator = new AtomicLong(0L);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestVolatileContentRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestVolatileContentRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestVolatileContentRepository.java index 5733164..feed31a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestVolatileContentRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestVolatileContentRepository.java @@ -16,8 +16,6 @@ */ package org.apache.nifi.controller.repository; -import org.apache.nifi.controller.repository.VolatileContentRepository; - import static org.junit.Assert.assertEquals; import java.io.ByteArrayOutputStream; @@ -25,6 +23,8 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -52,11 +52,11 @@ public class TestVolatileContentRepository { @Test public void testRedirects() throws IOException { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties"); - final NiFiProperties props = NiFiProperties.getInstance(); - props.setProperty(VolatileContentRepository.MAX_SIZE_PROPERTY, "10 MB"); - - final VolatileContentRepository contentRepo = new VolatileContentRepository(); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestVolatileContentRepository.class.getResource("/conf/nifi.properties").getFile()); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(VolatileContentRepository.MAX_SIZE_PROPERTY, "10 MB"); + final NiFiProperties nifiProps = NiFiProperties.createBasicNiFiProperties(null, addProps); + final VolatileContentRepository contentRepo = new VolatileContentRepository(nifiProps); contentRepo.initialize(claimManager); final ContentClaim claim = contentRepo.create(true); final OutputStream out = contentRepo.write(claim); @@ -106,11 +106,12 @@ public class TestVolatileContentRepository { @Test public void testMemoryIsFreed() throws IOException, InterruptedException { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties"); - final NiFiProperties props = NiFiProperties.getInstance(); - props.setProperty(VolatileContentRepository.MAX_SIZE_PROPERTY, "11 MB"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestVolatileContentRepository.class.getResource("/conf/nifi.properties").getFile()); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(VolatileContentRepository.MAX_SIZE_PROPERTY, "11 MB"); + final NiFiProperties nifiProps = NiFiProperties.createBasicNiFiProperties(null, addProps); + final VolatileContentRepository contentRepo = new VolatileContentRepository(nifiProps); - final VolatileContentRepository contentRepo = new VolatileContentRepository(); contentRepo.initialize(claimManager); final byte[] oneK = new byte[1024]; @@ -151,11 +152,11 @@ public class TestVolatileContentRepository { @Test public void testSimpleReadWrite() throws IOException { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties"); - final NiFiProperties props = NiFiProperties.getInstance(); - props.setProperty(VolatileContentRepository.MAX_SIZE_PROPERTY, "10 MB"); - - final VolatileContentRepository contentRepo = new VolatileContentRepository(); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestVolatileContentRepository.class.getResource("/conf/nifi.properties").getFile()); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(VolatileContentRepository.MAX_SIZE_PROPERTY, "11 MB"); + final NiFiProperties nifiProps = NiFiProperties.createBasicNiFiProperties(null, addProps); + final VolatileContentRepository contentRepo = new VolatileContentRepository(nifiProps); contentRepo.initialize(claimManager); final ContentClaim claim = contentRepo.create(true); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java index 51b654f..674f78f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java @@ -47,6 +47,7 @@ import org.apache.nifi.controller.repository.claim.StandardContentClaim; import org.apache.nifi.controller.repository.claim.StandardResourceClaimManager; import org.apache.nifi.controller.swap.StandardSwapContents; import org.apache.nifi.controller.swap.StandardSwapSummary; +import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.file.FileUtils; import org.junit.Before; import org.junit.BeforeClass; @@ -59,7 +60,7 @@ public class TestWriteAheadFlowFileRepository { @BeforeClass public static void setupProperties() { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestWriteAheadFlowFileRepository.class.getResource("/conf/nifi.properties").getFile()); } @Before @@ -95,26 +96,26 @@ public class TestWriteAheadFlowFileRepository { // Create a flowfile repo, update it once with a FlowFile that points to one resource claim. Then, // indicate that a FlowFile was swapped out. We should then be able to recover these FlowFiles and the // resource claims' counts should be updated for both the swapped out FlowFile and the non-swapped out FlowFile - try (final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository()) { + try (final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository(NiFiProperties.createBasicNiFiProperties(null, null))) { repo.initialize(claimManager); repo.loadFlowFiles(queueProvider, -1L); // Create a Repository Record that indicates that a FlowFile was created final FlowFileRecord flowFile1 = new StandardFlowFileRecord.Builder() - .id(1L) - .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") - .contentClaim(claim1) - .build(); + .id(1L) + .addAttribute("uuid", "11111111-1111-1111-1111-111111111111") + .contentClaim(claim1) + .build(); final StandardRepositoryRecord rec1 = new StandardRepositoryRecord(queue); rec1.setWorking(flowFile1); rec1.setDestination(queue); // Create a Record that we can swap out final FlowFileRecord flowFile2 = new StandardFlowFileRecord.Builder() - .id(2L) - .addAttribute("uuid", "11111111-1111-1111-1111-111111111112") - .contentClaim(claim2) - .build(); + .id(2L) + .addAttribute("uuid", "11111111-1111-1111-1111-111111111112") + .contentClaim(claim2) + .build(); final StandardRepositoryRecord rec2 = new StandardRepositoryRecord(queue); rec2.setWorking(flowFile2); @@ -129,9 +130,8 @@ public class TestWriteAheadFlowFileRepository { repo.swapFlowFilesOut(Collections.singletonList(flowFile2), queue, swapLocation); } - final ResourceClaimManager recoveryClaimManager = new StandardResourceClaimManager(); - try (final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository()) { + try (final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository(NiFiProperties.createBasicNiFiProperties(null, null))) { repo.initialize(recoveryClaimManager); final long largestId = repo.loadFlowFiles(queueProvider, 0L); @@ -162,7 +162,7 @@ public class TestWriteAheadFlowFileRepository { FileUtils.deleteFile(path.toFile(), true); } - final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository(); + final WriteAheadFlowFileRepository repo = new WriteAheadFlowFileRepository(NiFiProperties.createBasicNiFiProperties(null, null)); repo.initialize(new StandardResourceClaimManager()); final TestQueueProvider queueProvider = new TestQueueProvider(); @@ -216,7 +216,7 @@ public class TestWriteAheadFlowFileRepository { repo.close(); // restore - final WriteAheadFlowFileRepository repo2 = new WriteAheadFlowFileRepository(); + final WriteAheadFlowFileRepository repo2 = new WriteAheadFlowFileRepository(NiFiProperties.createBasicNiFiProperties(null, null)); repo2.initialize(new StandardResourceClaimManager()); repo2.loadFlowFiles(queueProvider, 0L); @@ -231,6 +231,7 @@ public class TestWriteAheadFlowFileRepository { } private static class TestQueueProvider implements QueueProvider { + private List<Connection> connectionList = new ArrayList<>(); public void addConnection(final Connection connection) { @@ -249,6 +250,7 @@ public class TestWriteAheadFlowFileRepository { } private static class MockFlowFileSwapManager implements FlowFileSwapManager { + private final Map<FlowFileQueue, Map<String, List<FlowFileRecord>>> swappedRecords = new HashMap<>(); @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java index 602efe7..2ebe8cf 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java @@ -56,8 +56,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.Random; import java.util.UUID; import java.util.concurrent.CountDownLatch; @@ -83,11 +85,7 @@ public class TestProcessorLifecycle { @Before public void before() throws Exception { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); - NiFiProperties.getInstance().setProperty(NiFiProperties.ADMINISTRATIVE_YIELD_DURATION, "1 sec"); - NiFiProperties.getInstance().setProperty(NiFiProperties.STATE_MANAGEMENT_CONFIG_FILE, "target/test-classes/state-management.xml"); - NiFiProperties.getInstance().setProperty(NiFiProperties.STATE_MANAGEMENT_LOCAL_PROVIDER_ID, "local-provider"); - fc = this.buildFlowControllerForTest(); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestProcessorLifecycle.class.getResource("/nifi.properties").getFile()); } @After @@ -99,6 +97,7 @@ public class TestProcessorLifecycle { @Test public void validateEnableOperation() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), @@ -117,9 +116,9 @@ public class TestProcessorLifecycle { assertEquals(ScheduledState.DISABLED, testProcNode.getPhysicalScheduledState()); } - @Test public void validateDisableOperation() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), @@ -145,6 +144,7 @@ public class TestProcessorLifecycle { */ @Test public void validateIdempotencyOfProcessorStartOperation() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -170,6 +170,7 @@ public class TestProcessorLifecycle { */ @Test public void validateStopCallsAreMeaninglessIfProcessorNotStarted() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -191,6 +192,7 @@ public class TestProcessorLifecycle { */ @Test public void validateSuccessfullAndOrderlyShutdown() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -232,6 +234,7 @@ public class TestProcessorLifecycle { */ @Test public void validateLifecycleOperationOrderWithConcurrentCallsToStartStop() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -287,6 +290,7 @@ public class TestProcessorLifecycle { */ @Test public void validateProcessorUnscheduledAndStoppedWhenStopIsCalledBeforeProcessorFullyStarted() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -316,11 +320,13 @@ public class TestProcessorLifecycle { } /** - * Validates that Processor is eventually started once invocation - * of @OnSchedule stopped throwing exceptions. + * Validates that Processor is eventually started once invocation of + * + * @OnSchedule stopped throwing exceptions. */ @Test public void validateProcessScheduledAfterAdministrativeDelayDueToTheOnScheduledException() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -352,6 +358,7 @@ public class TestProcessorLifecycle { */ @Test public void validateProcessorCanBeStoppedWhenOnScheduledConstantlyFails() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -382,7 +389,7 @@ public class TestProcessorLifecycle { */ @Test public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyInterruptable() throws Exception { - NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec"); + this.fc = buildFlowControllerForTest(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec"); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -410,7 +417,7 @@ public class TestProcessorLifecycle { */ @Test public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyUninterruptable() throws Exception { - NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec"); + this.fc = buildFlowControllerForTest(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec"); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -443,6 +450,7 @@ public class TestProcessorLifecycle { */ @Test public void validateProcessorCanBeStoppedWhenOnTriggerThrowsException() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -471,6 +479,7 @@ public class TestProcessorLifecycle { */ @Test(expected = IllegalStateException.class) public void validateStartFailsOnInvalidProcessorWithMissingProperty() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); @@ -485,13 +494,13 @@ public class TestProcessorLifecycle { */ @Test(expected = IllegalStateException.class) public void validateStartFailsOnInvalidProcessorWithDisabledService() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); ControllerServiceNode testServiceNode = fc.createControllerService(TestService.class.getName(), "serv", true); ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString()); - testProcNode.setProperty("P", "hello"); testProcNode.setProperty("S", testServiceNode.getIdentifier()); @@ -508,6 +517,7 @@ public class TestProcessorLifecycle { */ @Test public void validateStartSucceedsOnProcessorWithEnabledService() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); @@ -533,10 +543,12 @@ public class TestProcessorLifecycle { /** * Test deletion of processor when connected to another + * * @throws Exception exception */ @Test public void validateProcessorDeletion() throws Exception { + fc = this.buildFlowControllerForTest(); ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString()); this.setControllerRootGroup(fc, testGroup); @@ -652,18 +664,25 @@ public class TestProcessorLifecycle { testProcessor.setScenario(emptyRunnable, emptyRunnable, emptyRunnable, emptyRunnable); } - /** - * - */ - private FlowController buildFlowControllerForTest() throws Exception { - NiFiProperties properties = NiFiProperties.getInstance(); - properties.setProperty(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, MockProvenanceRepository.class.getName()); - properties.setProperty("nifi.remote.input.socket.port", ""); - properties.setProperty("nifi.remote.input.secure", ""); + private FlowController buildFlowControllerForTest(final String propKey, final String propValue) throws Exception { + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.ADMINISTRATIVE_YIELD_DURATION, "1 sec"); + addProps.put(NiFiProperties.STATE_MANAGEMENT_CONFIG_FILE, "target/test-classes/state-management.xml"); + addProps.put(NiFiProperties.STATE_MANAGEMENT_LOCAL_PROVIDER_ID, "local-provider"); + addProps.put(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, MockProvenanceRepository.class.getName()); + addProps.put("nifi.remote.input.socket.port", ""); + addProps.put("nifi.remote.input.secure", ""); + if (propKey != null && propValue != null) { + addProps.put(propKey, propValue); + } + final NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps); + return FlowController.createStandaloneInstance(mock(FlowFileEventRepository.class), nifiProperties, + mock(Authorizer.class), mock(AuditService.class), null, new VolatileBulletinRepository(), + new FileBasedVariableRegistry(nifiProperties.getVariableRegistryPropertiesPaths())); + } - return FlowController.createStandaloneInstance(mock(FlowFileEventRepository.class), properties, - mock(Authorizer.class), mock(AuditService.class), null, new VolatileBulletinRepository(), - new FileBasedVariableRegistry(properties.getVariableRegistryPropertiesPaths())); + private FlowController buildFlowControllerForTest() throws Exception { + return buildFlowControllerForTest(null, null); } /** @@ -683,6 +702,7 @@ public class TestProcessorLifecycle { /** */ public static class TestProcessor extends AbstractProcessor { + private Runnable onScheduleCallback; private Runnable onUnscheduleCallback; private Runnable onStopCallback; @@ -750,8 +770,8 @@ public class TestProcessorLifecycle { .identifiesControllerService(ITestservice.class) .build(); - return this.withService ? Arrays.asList(new PropertyDescriptor[] { PROP, SERVICE }) - : Arrays.asList(new PropertyDescriptor[] { PROP }); + return this.withService ? Arrays.asList(new PropertyDescriptor[]{PROP, SERVICE}) + : Arrays.asList(new PropertyDescriptor[]{PROP}); } @Override @@ -778,6 +798,7 @@ public class TestProcessorLifecycle { /** */ private static class EmptyRunnable implements Runnable { + @Override public void run() { @@ -787,6 +808,7 @@ public class TestProcessorLifecycle { /** */ private static class BlockingInterruptableRunnable implements Runnable { + @Override public void run() { try { @@ -800,6 +822,7 @@ public class TestProcessorLifecycle { /** */ private static class BlockingUninterruptableRunnable implements Runnable { + @Override public void run() { while (true) { @@ -815,6 +838,7 @@ public class TestProcessorLifecycle { /** */ private static class RandomOrFixedDelayedRunnable implements Runnable { + private final int delayLimit; private final boolean randomDelay; @@ -823,6 +847,7 @@ public class TestProcessorLifecycle { this.randomDelay = randomDelay; } Random random = new Random(); + @Override public void run() { try { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java index e50cc27..8a12e95 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; import java.util.Random; @@ -73,6 +72,7 @@ import org.junit.Test; import org.mockito.Mockito; public class TestStandardProcessScheduler { + private StandardProcessScheduler scheduler = null; private ReportingTaskNode taskNode = null; private TestReportingTask reportingTask = null; @@ -80,17 +80,18 @@ public class TestStandardProcessScheduler { private VariableRegistry variableRegistry = VariableRegistry.ENVIRONMENT_SYSTEM_REGISTRY; private FlowController controller; private ProcessGroup rootGroup; + private NiFiProperties nifiProperties; @Before public void setup() throws InitializationException { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); - this.refreshNiFiProperties(); - scheduler = new StandardProcessScheduler(Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider,variableRegistry); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestStandardProcessScheduler.class.getResource("/nifi.properties").getFile()); + this.nifiProperties = NiFiProperties.createBasicNiFiProperties(null, null); + scheduler = new StandardProcessScheduler(Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider, variableRegistry, nifiProperties); scheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, Mockito.mock(SchedulingAgent.class)); reportingTask = new TestReportingTask(); final ReportingInitializationContext config = new StandardReportingInitializationContext(UUID.randomUUID().toString(), "Test", SchedulingStrategy.TIMER_DRIVEN, "5 secs", - Mockito.mock(ComponentLog.class), null); + Mockito.mock(ComponentLog.class), null, nifiProperties); reportingTask.initialize(config); final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null, variableRegistry); @@ -102,10 +103,11 @@ public class TestStandardProcessScheduler { } /** - * We have run into an issue where a Reporting Task is scheduled to run but throws an Exception - * from a method with the @OnScheduled annotation. User stops Reporting Task, updates configuration - * to fix the issue. Reporting Task then finishes running @OnSchedule method and is then scheduled to run. - * This unit test is intended to verify that we have this resolved. + * We have run into an issue where a Reporting Task is scheduled to run but + * throws an Exception from a method with the @OnScheduled annotation. User + * stops Reporting Task, updates configuration to fix the issue. Reporting + * Task then finishes running @OnSchedule method and is then scheduled to + * run. This unit test is intended to verify that we have this resolved. */ @Test public void testReportingTaskDoesntKeepRunningAfterStop() throws InterruptedException, InitializationException { @@ -129,12 +131,13 @@ public class TestStandardProcessScheduler { public void testDisableControllerServiceWithProcessorTryingToStartUsingIt() throws InterruptedException { final Processor proc = new ServiceReferencingProcessor(); - final StandardControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(controller, scheduler, null, Mockito.mock(StateManagerProvider.class),variableRegistry); + final StandardControllerServiceProvider serviceProvider = + new StandardControllerServiceProvider(controller, scheduler, null, Mockito.mock(StateManagerProvider.class), variableRegistry, nifiProperties); final ControllerServiceNode service = serviceProvider.createControllerService(NoStartServiceImpl.class.getName(), "service", true); rootGroup.addControllerService(service); final ProcessorNode procNode = new StandardProcessorNode(proc, UUID.randomUUID().toString(), - new StandardValidationContextFactory(serviceProvider, variableRegistry), scheduler, serviceProvider); + new StandardValidationContextFactory(serviceProvider, variableRegistry), scheduler, serviceProvider, nifiProperties); rootGroup.addProcessor(procNode); procNode.setProperty(ServiceReferencingProcessor.SERVICE_DESC.getName(), service.getIdentifier()); @@ -150,12 +153,12 @@ public class TestStandardProcessScheduler { scheduler.disableControllerService(service); assertTrue(service.getState() == ControllerServiceState.DISABLING); assertFalse(service.isActive()); - Thread.sleep(1000); + Thread.sleep(2000); assertTrue(service.getState() == ControllerServiceState.DISABLED); } - private class TestReportingTask extends AbstractReportingTask { + private final AtomicBoolean failOnScheduled = new AtomicBoolean(true); private final AtomicInteger onScheduleAttempts = new AtomicInteger(0); private final AtomicInteger triggerCount = new AtomicInteger(0); @@ -175,8 +178,8 @@ public class TestStandardProcessScheduler { } } - private static class ServiceReferencingProcessor extends AbstractProcessor { + static final PropertyDescriptor SERVICE_DESC = new PropertyDescriptor.Builder() .name("service") .identifiesControllerService(NoStartService.class) @@ -195,15 +198,6 @@ public class TestStandardProcessScheduler { } } - private void refreshNiFiProperties() { - try { - final Field instanceField = NiFiProperties.class.getDeclaredField("instance"); - instanceField.setAccessible(true); - instanceField.set(null, null); - } catch (final Exception e) { - throw new IllegalStateException(e); - } - } /** * Validates the atomic nature of ControllerServiceNode.enable() method * which must only trigger @OnEnabled once, regardless of how many threads @@ -213,7 +207,7 @@ public class TestStandardProcessScheduler { @Test public void validateServiceEnablementLogicHappensOnlyOnce() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider,variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); assertFalse(serviceNode.isActive()); @@ -252,7 +246,7 @@ public class TestStandardProcessScheduler { @Test public void validateDisabledServiceCantBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); final SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation(); @@ -290,7 +284,7 @@ public class TestStandardProcessScheduler { @Test public void validateEnabledServiceCanOnlyBeDisabledOnce() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider,variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); final SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation(); @@ -324,7 +318,7 @@ public class TestStandardProcessScheduler { @Test public void validateDisablingOfTheFailedService() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(FailingService.class.getName(), "1", false); scheduler.enableControllerService(serviceNode); @@ -355,7 +349,7 @@ public class TestStandardProcessScheduler { @Test public void validateEnabledDisableMultiThread() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ExecutorService executor = Executors.newCachedThreadPool(); for (int i = 0; i < 200; i++) { final ControllerServiceNode serviceNode = provider @@ -398,7 +392,7 @@ public class TestStandardProcessScheduler { @Test public void validateNeverEnablingServiceCanStillBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider,variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(), "1", false); final LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation(); @@ -417,13 +411,14 @@ public class TestStandardProcessScheduler { /** * Validates that the service that is currently in ENABLING state can be - * disabled and that its @OnDisabled operation will be invoked as soon - * as @OnEnable finishes. + * disabled and that its @OnDisabled operation will be invoked as soon as + * + * @OnEnable finishes. */ @Test public void validateLongEnablingServiceCanStillBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider, variableRegistry, nifiProperties); final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(), "1", false); final LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation(); @@ -446,6 +441,7 @@ public class TestStandardProcessScheduler { } public static class FailingService extends AbstractControllerService { + @OnEnabled public void enable(final ConfigurationContext context) { throw new RuntimeException("intentional"); @@ -453,6 +449,7 @@ public class TestStandardProcessScheduler { } public static class RandomShortDelayEnablingService extends AbstractControllerService { + private final Random random = new Random(); @OnEnabled @@ -490,6 +487,7 @@ public class TestStandardProcessScheduler { } public static class LongEnablingService extends AbstractControllerService { + private final AtomicInteger enableCounter = new AtomicInteger(); private final AtomicInteger disableCounter = new AtomicInteger(); @@ -520,6 +518,6 @@ public class TestStandardProcessScheduler { } private ProcessScheduler createScheduler() { - return new StandardProcessScheduler(null, null, stateMgrProvider, variableRegistry); + return new StandardProcessScheduler(null, null, stateMgrProvider, variableRegistry, nifiProperties); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java index 77a1d8d..0e4571a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java @@ -19,7 +19,6 @@ package org.apache.nifi.controller.service; import org.apache.nifi.components.state.StateManager; import org.apache.nifi.components.state.StateManagerProvider; import org.apache.nifi.controller.ControllerService; -import org.apache.nifi.controller.StandardFlowServiceTest; import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.nar.NarClassLoaders; import org.apache.nifi.registry.VariableRegistry; @@ -36,14 +35,15 @@ public class StandardControllerServiceProviderTest { private ControllerService proxied; private ControllerService implementation; private static VariableRegistry variableRegistry; + private static NiFiProperties nifiProperties; @BeforeClass public static void setupSuite() throws Exception { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, StandardFlowServiceTest.class.getResource("/conf/nifi.properties").getFile()); - NiFiProperties properties = NiFiProperties.getInstance(); - NarClassLoaders.getInstance().init(properties.getFrameworkWorkingDirectory(), properties.getExtensionsWorkingDirectory()); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, StandardControllerServiceProviderTest.class.getResource("/conf/nifi.properties").getFile()); + nifiProperties = NiFiProperties.createBasicNiFiProperties(null, null); + NarClassLoaders.getInstance().init(nifiProperties.getFrameworkWorkingDirectory(), nifiProperties.getExtensionsWorkingDirectory()); ExtensionManager.discoverExtensions(NarClassLoaders.getInstance().getExtensionClassLoaders()); - variableRegistry = new FileBasedVariableRegistry(properties.getVariableRegistryPropertiesPaths()); + variableRegistry = new FileBasedVariableRegistry(nifiProperties.getVariableRegistryPropertiesPaths()); } @Before @@ -71,7 +71,7 @@ public class StandardControllerServiceProviderTest { @Override public void onComponentRemoved(String componentId) { } - }, variableRegistry); + }, variableRegistry, nifiProperties); ControllerServiceNode node = provider.createControllerService(clazz, id, true); proxied = node.getProxiedControllerService(); implementation = node.getControllerServiceImplementation(); @@ -96,4 +96,4 @@ public class StandardControllerServiceProviderTest { public void testCallImplementationInitialized() throws InitializationException { implementation.initialize(null); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java index c35fd59..3403346 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java @@ -47,12 +47,14 @@ import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.groups.StandardProcessGroup; import org.apache.nifi.processor.StandardValidationContextFactory; import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.util.NiFiProperties; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; import org.mockito.Mockito; public class TestStandardControllerServiceProvider { + private static StateManagerProvider stateManagerProvider = new StateManagerProvider() { @Override public StateManager getStateManager(final String componentId) { @@ -80,11 +82,11 @@ public class TestStandardControllerServiceProvider { @BeforeClass public static void setNiFiProps() { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestStandardControllerServiceProvider.class.getResource("/conf/nifi.properties").getFile()); } private StandardProcessScheduler createScheduler() { - return new StandardProcessScheduler(null, null, stateManagerProvider,variableRegistry); + return new StandardProcessScheduler(null, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); } @Test @@ -94,7 +96,8 @@ public class TestStandardControllerServiceProvider { Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider,variableRegistry); + final StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); final ControllerServiceNode serviceNode = provider.createControllerService(ServiceB.class.getName(), "B", false); provider.enableControllerService(serviceNode); @@ -108,7 +111,8 @@ public class TestStandardControllerServiceProvider { Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(group); final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider,variableRegistry); + final StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); final ControllerServiceNode serviceNodeB = provider.createControllerService(ServiceB.class.getName(), "B", false); final ControllerServiceNode serviceNodeA = provider.createControllerService(ServiceA.class.getName(), "A", false); @@ -167,7 +171,8 @@ public class TestStandardControllerServiceProvider { final FlowController controller = Mockito.mock(FlowController.class); Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider,variableRegistry); + final StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); // build a graph of controller services with dependencies as such: // @@ -212,14 +217,14 @@ public class TestStandardControllerServiceProvider { } } - @Test public void testOrderingOfServices() { final ProcessGroup procGroup = new MockProcessGroup(); final FlowController controller = Mockito.mock(FlowController.class); Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, null, null, stateManagerProvider,variableRegistry); + final StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, null, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); final ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false); final ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceB.class.getName(), "2", false); @@ -363,7 +368,7 @@ public class TestStandardControllerServiceProvider { private ProcessorNode createProcessor(final StandardProcessScheduler scheduler, final ControllerServiceProvider serviceProvider) { final ProcessorNode procNode = new StandardProcessorNode(new DummyProcessor(), UUID.randomUUID().toString(), - new StandardValidationContextFactory(serviceProvider, null), scheduler, serviceProvider); + new StandardValidationContextFactory(serviceProvider, null), scheduler, serviceProvider, NiFiProperties.createBasicNiFiProperties(null, null)); final ProcessGroup group = new StandardProcessGroup(UUID.randomUUID().toString(), serviceProvider, scheduler, null, null, null, variableRegistry); group.addProcessor(procNode); @@ -379,7 +384,8 @@ public class TestStandardControllerServiceProvider { Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); final StandardProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, null, null, stateManagerProvider,variableRegistry); + final StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, null, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); final ControllerServiceNode serviceNode = provider.createControllerService(ServiceA.class.getName(), "1", false); final ProcessorNode procNode = createProcessor(scheduler, provider); @@ -398,7 +404,8 @@ public class TestStandardControllerServiceProvider { public void validateEnableServices() { StandardProcessScheduler scheduler = createScheduler(); FlowController controller = Mockito.mock(FlowController.class); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider,variableRegistry); + StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); ProcessGroup procGroup = new MockProcessGroup(); Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); @@ -423,7 +430,7 @@ public class TestStandardControllerServiceProvider { E.setProperty(ServiceA.OTHER_SERVICE.getName(), "A"); E.setProperty(ServiceA.OTHER_SERVICE_2.getName(), "F"); - provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[] { A, B, C, D, E, F })); + provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[]{A, B, C, D, E, F})); assertTrue(A.isActive()); assertTrue(B.isActive()); @@ -443,7 +450,7 @@ public class TestStandardControllerServiceProvider { StandardProcessScheduler scheduler = createScheduler(); FlowController controller = Mockito.mock(FlowController.class); StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, - stateManagerProvider, variableRegistry); + stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); ProcessGroup procGroup = new MockProcessGroup(); Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); @@ -466,7 +473,7 @@ public class TestStandardControllerServiceProvider { F.setProperty(ServiceA.OTHER_SERVICE.getName(), "D"); D.setProperty(ServiceA.OTHER_SERVICE.getName(), "C"); - provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[] { C, F, A, B, D })); + provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[]{C, F, A, B, D})); assertTrue(A.isActive()); assertTrue(B.isActive()); @@ -479,7 +486,8 @@ public class TestStandardControllerServiceProvider { public void validateEnableServicesWithDisabledMissingService() { StandardProcessScheduler scheduler = createScheduler(); FlowController controller = Mockito.mock(FlowController.class); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider,variableRegistry); + StandardControllerServiceProvider provider = + new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider, variableRegistry, NiFiProperties.createBasicNiFiProperties(null, null)); ProcessGroup procGroup = new MockProcessGroup(); Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup); @@ -508,7 +516,7 @@ public class TestStandardControllerServiceProvider { serviceNode7.setProperty(ServiceC.REQ_SERVICE_2.getName(), "3"); provider.enableControllerServices(Arrays.asList( - new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5, serviceNode7})); + new ControllerServiceNode[]{serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5, serviceNode7})); assertFalse(serviceNode1.isActive()); assertFalse(serviceNode2.isActive()); assertFalse(serviceNode3.isActive()); @@ -518,7 +526,7 @@ public class TestStandardControllerServiceProvider { provider.enableControllerService(serviceNode6); provider.enableControllerServices(Arrays.asList( - new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5 })); + new ControllerServiceNode[]{serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5})); assertTrue(serviceNode1.isActive()); assertTrue(serviceNode2.isActive()); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java index 8ce97fa..f8e5b60 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.controller.state.providers.zookeeper; import java.io.IOException; @@ -38,8 +37,8 @@ import org.testng.Assert; public class TestZooKeeperStateProvider extends AbstractTestStateProvider { - private StateProvider provider; - private TestingServer zkServer; + private volatile StateProvider provider; + private volatile TestingServer zkServer; private static final Map<PropertyDescriptor, String> defaultProperties = new HashMap<>(); @@ -49,7 +48,6 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { defaultProperties.put(ZooKeeperStateProvider.ACCESS_CONTROL, ZooKeeperStateProvider.OPEN_TO_WORLD.getValue()); } - @Before public void setup() throws Exception { zkServer = new TestingServer(true); @@ -99,9 +97,11 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { @After public void clear() throws IOException { try { - getProvider().onComponentRemoved(componentId); - getProvider().disable(); - getProvider().shutdown(); + if (provider != null) { + provider.onComponentRemoved(componentId); + provider.disable(); + provider.shutdown(); + } } finally { if (zkServer != null) { zkServer.stop(); @@ -110,13 +110,11 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { } } - @Override protected StateProvider getProvider() { return provider; } - @Test(timeout = 20000) public void testStateTooLargeExceptionThrownOnSetState() throws InterruptedException { final Map<String, String> state = new HashMap<>(); @@ -152,7 +150,6 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { } } - @Test(timeout = 20000) public void testStateTooLargeExceptionThrownOnReplace() throws IOException, InterruptedException { final Map<String, String> state = new HashMap<>(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestContinuallyRunProcessorTask.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestContinuallyRunProcessorTask.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestContinuallyRunProcessorTask.java index 174e5fb..90e40a2 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestContinuallyRunProcessorTask.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestContinuallyRunProcessorTask.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.controller.tasks; import static org.junit.Assert.assertFalse; @@ -26,6 +25,7 @@ import java.util.Collections; import org.apache.nifi.connectable.Connection; import org.apache.nifi.controller.ProcessorNode; import org.apache.nifi.controller.queue.FlowFileQueue; +import org.apache.nifi.util.NiFiProperties; import org.junit.Test; import org.mockito.Mockito; @@ -33,7 +33,7 @@ public class TestContinuallyRunProcessorTask { @Test public void testIsWorkToDo() { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestContinuallyRunProcessorTask.class.getResource("/conf/nifi.properties").getFile()); final ProcessorNode procNode = Mockito.mock(ProcessorNode.class); Mockito.when(procNode.hasIncomingConnection()).thenReturn(false); @@ -60,7 +60,6 @@ public class TestContinuallyRunProcessorTask { when(selfLoopingConnection.getFlowFileQueue()).thenReturn(nonEmptyQueue); assertTrue(ContinuallyRunProcessorTask.isWorkToDo(procNode)); - // Test with only a non-looping Connection that has no FlowFiles final Connection emptyConnection = Mockito.mock(Connection.class); when(emptyConnection.getSource()).thenReturn(Mockito.mock(ProcessorNode.class));
