http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/VolatileContentRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/VolatileContentRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/VolatileContentRepository.java index 08b7e80..95b503b 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/VolatileContentRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/VolatileContentRepository.java @@ -56,24 +56,33 @@ import org.slf4j.LoggerFactory; /** * <p> - * An in-memory implementation of the {@link ContentRepository} interface. This implementation stores FlowFile content in the Java heap and keeps track of the number of bytes used. If the number of - * bytes used by FlowFile content exceeds some threshold (configured via the <code>nifi.volatile.content.repository.max.size</code> property in the NiFi properties with a default of 100 MB), one of - * two situations will occur: + * An in-memory implementation of the {@link ContentRepository} interface. This + * implementation stores FlowFile content in the Java heap and keeps track of + * the number of bytes used. If the number of bytes used by FlowFile content + * exceeds some threshold (configured via the + * <code>nifi.volatile.content.repository.max.size</code> property in the NiFi + * properties with a default of 100 MB), one of two situations will occur: * </p> * * <ul> - * <li><b>Backup Repository:</b> If a Backup Repository has been specified (via the {@link #setBackupRepository(ContentRepository)} method), the content will be stored in the backup repository and all - * access to the FlowFile content will automatically and transparently be proxied to the backup repository. + * <li><b>Backup Repository:</b> If a Backup Repository has been specified (via + * the {@link #setBackupRepository(ContentRepository)} method), the content will + * be stored in the backup repository and all access to the FlowFile content + * will automatically and transparently be proxied to the backup repository. * </li> * <li> - * <b>Without Backup Repository:</b> If no Backup Repository has been specified, when the threshold is exceeded, an IOException will be thrown. + * <b>Without Backup Repository:</b> If no Backup Repository has been specified, + * when the threshold is exceeded, an IOException will be thrown. * </li> * </ul> * * <p> - * When a Content Claim is created via the {@link #create(boolean)} method, if the <code>lossTolerant</code> flag is set to <code>false</code>, the Backup Repository will be used to create the Content - * Claim and any accesses to the ContentClaim will be proxied to the Backup Repository. If the Backup Repository has not been specified, attempting to create a non-loss-tolerant ContentClaim will - * result in an {@link IllegalStateException} being thrown. + * When a Content Claim is created via the {@link #create(boolean)} method, if + * the <code>lossTolerant</code> flag is set to <code>false</code>, the Backup + * Repository will be used to create the Content Claim and any accesses to the + * ContentClaim will be proxied to the Backup Repository. If the Backup + * Repository has not been specified, attempting to create a non-loss-tolerant + * ContentClaim will result in an {@link IllegalStateException} being thrown. * </p> */ public class VolatileContentRepository implements ContentRepository { @@ -98,13 +107,17 @@ public class VolatileContentRepository implements ContentRepository { private ResourceClaimManager claimManager; // effectively final + /** + * Default no args constructor for service loading only + */ public VolatileContentRepository() { - this(NiFiProperties.getInstance()); + maxBytes = 0; + memoryManager = null; } - public VolatileContentRepository(final NiFiProperties properties) { - final String maxSize = properties.getProperty(MAX_SIZE_PROPERTY); - final String blockSizeVal = properties.getProperty(BLOCK_SIZE_PROPERTY); + public VolatileContentRepository(final NiFiProperties nifiProperties) { + final String maxSize = nifiProperties.getProperty(MAX_SIZE_PROPERTY); + final String blockSizeVal = nifiProperties.getProperty(BLOCK_SIZE_PROPERTY); if (maxSize == null) { maxBytes = (long) DataUnit.B.convert(100D, DataUnit.MB); @@ -137,7 +150,8 @@ public class VolatileContentRepository implements ContentRepository { } /** - * Specifies a Backup Repository where data should be written if this Repository fills up + * Specifies a Backup Repository where data should be written if this + * Repository fills up * * @param backup repo backup */ @@ -388,7 +402,7 @@ public class VolatileContentRepository implements ContentRepository { final StandardOpenOption openOption = append ? StandardOpenOption.APPEND : StandardOpenOption.CREATE; try (final InputStream in = read(claim); - final OutputStream destinationStream = Files.newOutputStream(destination, openOption)) { + final OutputStream destinationStream = Files.newOutputStream(destination, openOption)) { if (offset > 0) { StreamUtils.skip(in, offset);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java index 2e5f005..36d592c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java @@ -63,12 +63,21 @@ import org.wali.WriteAheadRepository; * </p> * * <p> - * We expose a property named <code>nifi.flowfile.repository.always.sync</code> that is a boolean value indicating whether or not to force WALI to sync with disk on each update. By default, the value - * is <code>false</code>. This is needed only in situations in which power loss is expected and not mitigated by Uninterruptable Power Sources (UPS) or when running in an unstable Virtual Machine for - * instance. Otherwise, we will flush the data that is written to the Operating System and the Operating System will be responsible to flush its buffers when appropriate. The Operating System can be - * configured to hold only a certain buffer size or not to buffer at all, as well. When using a UPS, this is generally not an issue, as the machine is typically notified before dying, in which case - * the Operating System will flush the data to disk. Additionally, most disks on enterprise servers also have battery backups that can power the disks long enough to flush their buffers. For this - * reason, we choose instead to not sync to disk for every write but instead sync only when we checkpoint. + * We expose a property named <code>nifi.flowfile.repository.always.sync</code> + * that is a boolean value indicating whether or not to force WALI to sync with + * disk on each update. By default, the value is <code>false</code>. This is + * needed only in situations in which power loss is expected and not mitigated + * by Uninterruptable Power Sources (UPS) or when running in an unstable Virtual + * Machine for instance. Otherwise, we will flush the data that is written to + * the Operating System and the Operating System will be responsible to flush + * its buffers when appropriate. The Operating System can be configured to hold + * only a certain buffer size or not to buffer at all, as well. When using a + * UPS, this is generally not an issue, as the machine is typically notified + * before dying, in which case the Operating System will flush the data to disk. + * Additionally, most disks on enterprise servers also have battery backups that + * can power the disks long enough to flush their buffers. For this reason, we + * choose instead to not sync to disk for every write but instead sync only when + * we checkpoint. * </p> */ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncListener { @@ -112,15 +121,24 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis // on restart. private final ConcurrentMap<Integer, BlockingQueue<ResourceClaim>> claimsAwaitingDestruction = new ConcurrentHashMap<>(); + /** + * default no args constructor for service loading only. + */ public WriteAheadFlowFileRepository() { - final NiFiProperties properties = NiFiProperties.getInstance(); + alwaysSync = false; + checkpointDelayMillis = 0l; + flowFileRepositoryPath = null; + numPartitions = 0; + checkpointExecutor = null; + } - alwaysSync = Boolean.parseBoolean(properties.getProperty(NiFiProperties.FLOWFILE_REPOSITORY_ALWAYS_SYNC, "false")); + public WriteAheadFlowFileRepository(final NiFiProperties nifiProperties) { + alwaysSync = Boolean.parseBoolean(nifiProperties.getProperty(NiFiProperties.FLOWFILE_REPOSITORY_ALWAYS_SYNC, "false")); // determine the database file path and ensure it exists - flowFileRepositoryPath = properties.getFlowFileRepositoryPath(); - numPartitions = properties.getFlowFileRepositoryPartitions(); - checkpointDelayMillis = FormatUtils.getTimeDuration(properties.getFlowFileRepositoryCheckpointInterval(), TimeUnit.MILLISECONDS); + flowFileRepositoryPath = nifiProperties.getFlowFileRepositoryPath(); + numPartitions = nifiProperties.getFlowFileRepositoryPartitions(); + checkpointDelayMillis = FormatUtils.getTimeDuration(nifiProperties.getFlowFileRepositoryCheckpointInterval(), TimeUnit.MILLISECONDS); checkpointExecutor = Executors.newSingleThreadScheduledExecutor(); } @@ -253,7 +271,6 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis } } - @Override public void onSync(final int partitionIndex) { final BlockingQueue<ResourceClaim> claimQueue = claimsAwaitingDestruction.get(Integer.valueOf(partitionIndex)); @@ -282,7 +299,9 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis } /** - * Swaps the FlowFiles that live on the given Connection out to disk, using the specified Swap File and returns the number of FlowFiles that were persisted. + * Swaps the FlowFiles that live on the given Connection out to disk, using + * the specified Swap File and returns the number of FlowFiles that were + * persisted. * * @param queue queue to swap out * @param swapLocation location to swap to @@ -397,6 +416,7 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis } private static class WriteAheadRecordSerde implements SerDe<RepositoryRecord> { + private static final int CURRENT_ENCODING_VERSION = 9; public static final byte ACTION_CREATE = 0; @@ -551,7 +571,7 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis if (version > 1) { // read the lineage identifiers and lineage start date, which were added in version 2. - if(version < 9){ + if (version < 9) { final int numLineageIds = in.readInt(); for (int i = 0; i < numLineageIds; i++) { in.readUTF(); //skip identifiers @@ -662,7 +682,7 @@ public class WriteAheadFlowFileRepository implements FlowFileRepository, SyncLis if (version > 1) { // read the lineage identifiers and lineage start date, which were added in version 2. - if(version < 9) { + if (version < 9) { final int numLineageIds = in.readInt(); for (int i = 0; i < numLineageIds; i++) { in.readUTF(); //skip identifiers http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java index aadbff0..10a746e 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java @@ -62,7 +62,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Responsible for scheduling Processors, Ports, and Funnels to run at regular intervals + * Responsible for scheduling Processors, Ports, and Funnels to run at regular + * intervals */ public final class StandardProcessScheduler implements ProcessScheduler { @@ -84,14 +85,19 @@ public final class StandardProcessScheduler implements ProcessScheduler { private final StringEncryptor encryptor; private final VariableRegistry variableRegistry; - public StandardProcessScheduler(final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, - final StateManagerProvider stateManagerProvider, final VariableRegistry variableRegistry) { + public StandardProcessScheduler( + final ControllerServiceProvider controllerServiceProvider, + final StringEncryptor encryptor, + final StateManagerProvider stateManagerProvider, + final VariableRegistry variableRegistry, + final NiFiProperties nifiProperties + ) { this.controllerServiceProvider = controllerServiceProvider; this.encryptor = encryptor; this.stateManagerProvider = stateManagerProvider; this.variableRegistry = variableRegistry; - administrativeYieldDuration = NiFiProperties.getInstance().getAdministrativeYieldDuration(); + administrativeYieldDuration = nifiProperties.getAdministrativeYieldDuration(); administrativeYieldMillis = FormatUtils.getTimeDuration(administrativeYieldDuration, TimeUnit.MILLISECONDS); frameworkTaskExecutor = new FlowEngine(4, "Framework Task Thread"); @@ -216,8 +222,8 @@ public final class StandardProcessScheduler implements ProcessScheduler { componentLog.error("Failed to invoke @OnEnabled method due to {}", cause); LOG.error("Failed to invoke the On-Scheduled Lifecycle methods of {} due to {}; administratively yielding this " - + "ReportingTask and will attempt to schedule it again after {}", - new Object[] { reportingTask, e.toString(), administrativeYieldDuration }, e); + + "ReportingTask and will attempt to schedule it again after {}", + new Object[]{reportingTask, e.toString(), administrativeYieldDuration}, e); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, reportingTask, taskNode.getConfigurationContext()); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, reportingTask, taskNode.getConfigurationContext()); @@ -265,7 +271,7 @@ public final class StandardProcessScheduler implements ProcessScheduler { componentLog.error("Failed to invoke @OnUnscheduled method due to {}", cause); LOG.error("Failed to invoke the @OnUnscheduled methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}", - reportingTask, cause.toString(), administrativeYieldDuration); + reportingTask, cause.toString(), administrativeYieldDuration); LOG.error("", cause); try { @@ -290,8 +296,9 @@ public final class StandardProcessScheduler implements ProcessScheduler { * Starts the given {@link Processor} by invoking its * {@link ProcessorNode#start(ScheduledExecutorService, long, org.apache.nifi.processor.ProcessContext, Runnable)} * . + * * @see StandardProcessorNode#start(ScheduledExecutorService, long, - * org.apache.nifi.processor.ProcessContext, Runnable). + * org.apache.nifi.processor.ProcessContext, Runnable). */ @Override public synchronized void startProcessor(final ProcessorNode procNode) { @@ -324,8 +331,9 @@ public final class StandardProcessScheduler implements ProcessScheduler { * Stops the given {@link Processor} by invoking its * {@link ProcessorNode#stop(ScheduledExecutorService, org.apache.nifi.processor.ProcessContext, Callable)} * . + * * @see StandardProcessorNode#stop(ScheduledExecutorService, - * org.apache.nifi.processor.ProcessContext, Callable) + * org.apache.nifi.processor.ProcessContext, Callable) */ @Override public synchronized void stopProcessor(final ProcessorNode procNode) { @@ -524,8 +532,7 @@ public final class StandardProcessScheduler implements ProcessScheduler { * no ScheduleState current is registered, one is created and registered * atomically, and then that value is returned. * - * @param schedulable - * schedulable + * @param schedulable schedulable * @return scheduled state */ private ScheduleState getScheduleState(final Object schedulable) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java index f94beff..fcd901f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java @@ -55,15 +55,20 @@ public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent { private volatile String adminYieldDuration = "1 sec"; - public TimerDrivenSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor encryptor, - final VariableRegistry variableRegistry) { + public TimerDrivenSchedulingAgent( + final FlowController flowController, + final FlowEngine flowEngine, + final ProcessContextFactory contextFactory, + final StringEncryptor encryptor, + final VariableRegistry variableRegistry, + final NiFiProperties nifiProperties) { super(flowEngine); this.flowController = flowController; this.contextFactory = contextFactory; this.encryptor = encryptor; this.variableRegistry = variableRegistry; - final String boredYieldDuration = NiFiProperties.getInstance().getBoredYieldDuration(); + final String boredYieldDuration = nifiProperties.getBoredYieldDuration(); try { noWorkYieldNanos = FormatUtils.getTimeDuration(boredYieldDuration, TimeUnit.NANOSECONDS); } catch (final IllegalArgumentException e) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java index 71cd793..59e668c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java @@ -16,6 +16,7 @@ */ package org.apache.nifi.controller.service; +import java.io.File; import java.util.Set; import org.apache.nifi.components.state.StateManager; @@ -23,6 +24,7 @@ import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.ControllerServiceInitializationContext; import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.util.NiFiProperties; public class StandardControllerServiceInitializationContext implements ControllerServiceInitializationContext, ControllerServiceLookup { @@ -30,12 +32,17 @@ public class StandardControllerServiceInitializationContext implements Controlle private final ControllerServiceProvider serviceProvider; private final ComponentLog logger; private final StateManager stateManager; + private final NiFiProperties nifiProperties; - public StandardControllerServiceInitializationContext(final String identifier, final ComponentLog logger, final ControllerServiceProvider serviceProvider, final StateManager stateManager) { + public StandardControllerServiceInitializationContext( + final String identifier, final ComponentLog logger, + final ControllerServiceProvider serviceProvider, final StateManager stateManager, + final NiFiProperties nifiProperties) { this.id = identifier; this.logger = logger; this.serviceProvider = serviceProvider; this.stateManager = stateManager; + this.nifiProperties = nifiProperties; } @Override @@ -87,4 +94,19 @@ public class StandardControllerServiceInitializationContext implements Controlle public StateManager getStateManager() { return stateManager; } + + @Override + public String getKerberosServicePrincipal() { + return nifiProperties.getKerberosServicePrincipal(); + } + + @Override + public File getKerberosServiceKeytab() { + return new File(nifiProperties.getKerberosKeytabLocation()); + } + + @Override + public File getKerberosConfigurationFile() { + return nifiProperties.getKerberosConfigurationFile(); + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java index 9065264..9fd1ca2 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java @@ -59,6 +59,7 @@ import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.BulletinRepository; import org.apache.nifi.reporting.Severity; +import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.ReflectionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -73,6 +74,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi private final StateManagerProvider stateManagerProvider; private final VariableRegistry variableRegistry; private final FlowController flowController; + private final NiFiProperties nifiProperties; static { // methods that are okay to be called when the service is disabled. @@ -87,13 +89,14 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi } public StandardControllerServiceProvider(final FlowController flowController, final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, - final StateManagerProvider stateManagerProvider, final VariableRegistry variableRegistry) { + final StateManagerProvider stateManagerProvider, final VariableRegistry variableRegistry, final NiFiProperties nifiProperties) { this.flowController = flowController; this.processScheduler = scheduler; this.bulletinRepo = bulletinRepo; this.stateManagerProvider = stateManagerProvider; this.variableRegistry = variableRegistry; + this.nifiProperties = nifiProperties; } private Class<?>[] getInterfaces(final Class<?> cls) { @@ -189,7 +192,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi logger.info("Created Controller Service of type {} with identifier {}", type, id); final ComponentLog serviceLogger = new SimpleProcessLogger(id, originalService); - originalService.initialize(new StandardControllerServiceInitializationContext(id, serviceLogger, this, getStateManager(id))); + originalService.initialize(new StandardControllerServiceInitializationContext(id, serviceLogger, this, getStateManager(id), nifiProperties)); final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this, variableRegistry); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java index 828bdfe..4c620d1 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java @@ -40,10 +40,15 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit private volatile long lastCaptureTime = 0L; + /** + * Default no args constructor for service loading only + */ + public VolatileComponentStatusRepository(){ + captures = null; + } - public VolatileComponentStatusRepository() { - final NiFiProperties properties = NiFiProperties.getInstance(); - final int numDataPoints = properties.getIntegerProperty(NUM_DATA_POINTS_PROPERTY, DEFAULT_NUM_DATA_POINTS); + public VolatileComponentStatusRepository(final NiFiProperties nifiProperties) { + final int numDataPoints = nifiProperties.getIntegerProperty(NUM_DATA_POINTS_PROPERTY, DEFAULT_NUM_DATA_POINTS); captures = new RingBuffer<>(numDataPoints); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/encrypt/StringEncryptor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/encrypt/StringEncryptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/encrypt/StringEncryptor.java index 5de1beb..15043c3 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/encrypt/StringEncryptor.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/encrypt/StringEncryptor.java @@ -26,14 +26,16 @@ import org.jasypt.exceptions.EncryptionOperationNotPossibleException; /** * <p> - * An application specific string encryptor that collects configuration from the application properties, system properties, and/or system environment. + * An application specific string encryptor that collects configuration from the + * application properties, system properties, and/or system environment. * </p> * * <p> * Instance of this class are thread-safe</p> * * <p> - * The encryption provider and algorithm is configured using the application properties: + * The encryption provider and algorithm is configured using the application + * properties: * <ul> * <li>nifi.sensitive.props.provider</li> * <li>nifi.sensitive.props.algorithm</li> @@ -71,18 +73,21 @@ public final class StringEncryptor { } /** - * Creates an instance of the nifi sensitive property encryptor. Validates that the encryptor is actually working. + * Creates an instance of the nifi sensitive property encryptor. Validates + * that the encryptor is actually working. * + * @param niFiProperties properties * @return encryptor - * @throws EncryptionException if any issues arise initializing or validating the encryptor + * @throws EncryptionException if any issues arise initializing or + * validating the encryptor */ - public static StringEncryptor createEncryptor() throws EncryptionException { + public static StringEncryptor createEncryptor(final NiFiProperties niFiProperties) throws EncryptionException { Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider()); - final String sensitivePropAlgorithmVal = NiFiProperties.getInstance().getProperty(NF_SENSITIVE_PROPS_ALGORITHM); - final String sensitivePropProviderVal = NiFiProperties.getInstance().getProperty(NF_SENSITIVE_PROPS_PROVIDER); - final String sensitivePropValueNifiPropVar = NiFiProperties.getInstance().getProperty(NF_SENSITIVE_PROPS_KEY, DEFAULT_SENSITIVE_PROPS_KEY); + final String sensitivePropAlgorithmVal = niFiProperties.getProperty(NF_SENSITIVE_PROPS_ALGORITHM); + final String sensitivePropProviderVal = niFiProperties.getProperty(NF_SENSITIVE_PROPS_PROVIDER); + final String sensitivePropValueNifiPropVar = niFiProperties.getProperty(NF_SENSITIVE_PROPS_KEY, DEFAULT_SENSITIVE_PROPS_KEY); if (StringUtils.isBlank(sensitivePropAlgorithmVal)) { throw new EncryptionException(NF_SENSITIVE_PROPS_ALGORITHM + "must bet set"); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java index f73dce5..5eceb52 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java @@ -45,10 +45,12 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD private final Path flowXmlPath; private final StringEncryptor encryptor; private final FlowConfigurationArchiveManager archiveManager; + private final NiFiProperties nifiProperties; private static final Logger LOG = LoggerFactory.getLogger(StandardXMLFlowConfigurationDAO.class); - public StandardXMLFlowConfigurationDAO(final Path flowXml, final StringEncryptor encryptor) throws IOException { + public StandardXMLFlowConfigurationDAO(final Path flowXml, final StringEncryptor encryptor, final NiFiProperties nifiProperties) throws IOException { + this.nifiProperties = nifiProperties; final File flowXmlFile = flowXml.toFile(); if (!flowXmlFile.exists()) { // createDirectories would throw an exception if the directory exists but is a symbolic link @@ -64,7 +66,7 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD this.flowXmlPath = flowXml; this.encryptor = encryptor; - this.archiveManager = new FlowConfigurationArchiveManager(flowXmlPath, NiFiProperties.getInstance()); + this.archiveManager = new FlowConfigurationArchiveManager(flowXmlPath, nifiProperties); } @Override @@ -77,7 +79,7 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD public synchronized void load(final FlowController controller, final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { - final FlowSynchronizer flowSynchronizer = new StandardFlowSynchronizer(encryptor); + final FlowSynchronizer flowSynchronizer = new StandardFlowSynchronizer(encryptor, nifiProperties); controller.synchronize(flowSynchronizer, dataFlow); if (StandardFlowSynchronizer.isEmpty(dataFlow)) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessorInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessorInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessorInitializationContext.java index 8855545..3138fe9 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessorInitializationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessorInitializationContext.java @@ -16,10 +16,12 @@ */ package org.apache.nifi.processor; +import java.io.File; import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.controller.NodeTypeProvider; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.util.NiFiProperties; public class StandardProcessorInitializationContext implements ProcessorInitializationContext { @@ -27,12 +29,17 @@ public class StandardProcessorInitializationContext implements ProcessorInitiali private final ComponentLog logger; private final ControllerServiceProvider serviceProvider; private final NodeTypeProvider nodeTypeProvider; + private final NiFiProperties nifiProperties; - public StandardProcessorInitializationContext(final String identifier, final ComponentLog componentLog, final ControllerServiceProvider serviceProvider, NodeTypeProvider nodeTypeProvider) { + public StandardProcessorInitializationContext( + final String identifier, final ComponentLog componentLog, + final ControllerServiceProvider serviceProvider, final NodeTypeProvider nodeTypeProvider, + final NiFiProperties nifiProperties) { this.identifier = identifier; this.logger = componentLog; this.serviceProvider = serviceProvider; this.nodeTypeProvider = nodeTypeProvider; + this.nifiProperties = nifiProperties; } @Override @@ -54,4 +61,19 @@ public class StandardProcessorInitializationContext implements ProcessorInitiali public NodeTypeProvider getNodeTypeProvider() { return nodeTypeProvider; } + + @Override + public String getKerberosServicePrincipal() { + return nifiProperties.getKerberosServicePrincipal(); + } + + @Override + public File getKerberosServiceKeytab() { + return new File(nifiProperties.getKerberosKeytabLocation()); + } + + @Override + public File getKerberosConfigurationFile() { + return nifiProperties.getKerberosConfigurationFile(); + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java index 43947ed..1c9d182 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java @@ -76,8 +76,9 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; import static java.util.Objects.requireNonNull; /** - * Represents the Root Process Group of a remote NiFi Instance. Holds information about that remote instance, as well as {@link IncomingPort}s and {@link OutgoingPort}s for communicating with the - * remote instance. + * Represents the Root Process Group of a remote NiFi Instance. Holds + * information about that remote instance, as well as {@link IncomingPort}s and + * {@link OutgoingPort}s for communicating with the remote instance. */ public class StandardRemoteProcessGroup implements RemoteProcessGroup { @@ -95,6 +96,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { private final String protocol; private final ProcessScheduler scheduler; private final EventReporter eventReporter; + private final NiFiProperties nifiProperties; private final AtomicReference<String> name = new AtomicReference<>(); private final AtomicReference<Position> position = new AtomicReference<>(); @@ -115,7 +117,6 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { private volatile String proxyUser; private volatile String proxyPassword; - private final ReadWriteLock rwLock = new ReentrantReadWriteLock(); private final Lock readLock = rwLock.readLock(); private final Lock writeLock = rwLock.writeLock(); @@ -137,7 +138,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { private final ScheduledExecutorService backgroundThreadExecutor; public StandardRemoteProcessGroup(final String id, final String targetUri, final ProcessGroup processGroup, - final FlowController flowController, final SSLContext sslContext) { + final FlowController flowController, final SSLContext sslContext, final NiFiProperties nifiProperties) { + this.nifiProperties = nifiProperties; this.id = requireNonNull(id); this.flowController = requireNonNull(flowController); final URI uri; @@ -174,7 +176,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { final String sourceId = StandardRemoteProcessGroup.this.getIdentifier(); final String sourceName = StandardRemoteProcessGroup.this.getName(); bulletinRepository.addBulletin(BulletinFactory.createBulletin(groupId, sourceId, ComponentType.REMOTE_PROCESS_GROUP, - sourceName, category, severity.name(), message)); + sourceName, category, severity.name(), message)); } }; @@ -400,8 +402,11 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * Changes the currently configured input ports to the ports described in the given set. If any port is currently configured that is not in the set given, that port will be shutdown and removed. - * If any port is currently not configured and is in the set given, that port will be instantiated and started. + * Changes the currently configured input ports to the ports described in + * the given set. If any port is currently configured that is not in the set + * given, that port will be shutdown and removed. If any port is currently + * not configured and is in the set given, that port will be instantiated + * and started. * * @param ports the new ports * @@ -450,10 +455,12 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * Returns a boolean indicating whether or not an Output Port exists with the given ID + * Returns a boolean indicating whether or not an Output Port exists with + * the given ID * * @param id identifier of port - * @return <code>true</code> if an Output Port exists with the given ID, <code>false</code> otherwise. + * @return <code>true</code> if an Output Port exists with the given ID, + * <code>false</code> otherwise. */ public boolean containsOutputPort(final String id) { readLock.lock(); @@ -465,8 +472,11 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * Changes the currently configured output ports to the ports described in the given set. If any port is currently configured that is not in the set given, that port will be shutdown and removed. - * If any port is currently not configured and is in the set given, that port will be instantiated and started. + * Changes the currently configured output ports to the ports described in + * the given set. If any port is currently configured that is not in the set + * given, that port will be shutdown and removed. If any port is currently + * not configured and is in the set given, that port will be instantiated + * and started. * * @param ports the new ports * @@ -519,7 +529,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { * * * @throws NullPointerException if the given output Port is null - * @throws IllegalStateException if the port does not belong to this remote process group + * @throws IllegalStateException if the port does not belong to this remote + * process group */ @Override public void removeNonExistentPort(final RemoteGroupPort port) { @@ -597,11 +608,13 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * Adds an Output Port to this Remote Process Group that is described by this DTO. + * Adds an Output Port to this Remote Process Group that is described by + * this DTO. * * @param descriptor * - * @throws IllegalStateException if an Output Port already exists with the ID given by dto.getId() + * @throws IllegalStateException if an Output Port already exists with the + * ID given by dto.getId() */ private void addOutputPort(final RemoteProcessGroupPortDescriptor descriptor) { writeLock.lock(); @@ -611,7 +624,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(), - this, TransferDirection.RECEIVE, ConnectableType.REMOTE_OUTPUT_PORT, sslContext, scheduler); + this, TransferDirection.RECEIVE, ConnectableType.REMOTE_OUTPUT_PORT, sslContext, scheduler, nifiProperties); outputPorts.put(descriptor.getId(), port); if (descriptor.getConcurrentlySchedulableTaskCount() != null) { @@ -627,7 +640,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { /** * @param portIdentifier the ID of the Port to send FlowFiles to - * @return {@link RemoteGroupPort} that can be used to send FlowFiles to the port whose ID is given on the remote instance + * @return {@link RemoteGroupPort} that can be used to send FlowFiles to the + * port whose ID is given on the remote instance */ @Override public RemoteGroupPort getInputPort(final String portIdentifier) { @@ -644,7 +658,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * @return a set of {@link OutgoingPort}s used for transmitting FlowFiles to the remote instance + * @return a set of {@link OutgoingPort}s used for transmitting FlowFiles to + * the remote instance */ @Override public Set<RemoteGroupPort> getInputPorts() { @@ -659,11 +674,13 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * Adds an InputPort to this ProcessGroup that is described by the given DTO. + * Adds an InputPort to this ProcessGroup that is described by the given + * DTO. * * @param descriptor port descriptor * - * @throws IllegalStateException if an Input Port already exists with the ID given by the ID of the DTO. + * @throws IllegalStateException if an Input Port already exists with the ID + * given by the ID of the DTO. */ private void addInputPort(final RemoteProcessGroupPortDescriptor descriptor) { writeLock.lock(); @@ -673,7 +690,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(), this, - TransferDirection.SEND, ConnectableType.REMOTE_INPUT_PORT, sslContext, scheduler); + TransferDirection.SEND, ConnectableType.REMOTE_INPUT_PORT, sslContext, scheduler, nifiProperties); if (descriptor.getConcurrentlySchedulableTaskCount() != null) { port.setMaxConcurrentTasks(descriptor.getConcurrentlySchedulableTaskCount()); @@ -703,7 +720,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } /** - * @return a set of {@link RemoteGroupPort}s used for receiving FlowFiles from the remote instance + * @return a set of {@link RemoteGroupPort}s used for receiving FlowFiles + * from the remote instance */ @Override public Set<RemoteGroupPort> getOutputPorts() { @@ -772,10 +790,9 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { writeLock.lock(); try { - final NiFiProperties props = NiFiProperties.getInstance(); - this.destinationSecure = props.isSiteToSiteSecure(); - this.listeningPort = props.getRemoteInputPort(); - this.listeningHttpPort = props.getRemoteInputHttpPort(); + this.destinationSecure = nifiProperties.isSiteToSiteSecure(); + this.listeningPort = nifiProperties.getRemoteInputPort(); + this.listeningHttpPort = nifiProperties.getRemoteInputHttpPort(); refreshContentsTimestamp = System.currentTimeMillis(); } finally { @@ -815,8 +832,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { // perform the request final ControllerDTO dto; try ( - final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient(); - ){ + final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient();) { dto = apiClient.getController(); } catch (IOException e) { writeLock.lock(); @@ -1138,7 +1154,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { @Override public void run() { - try (final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient()){ + try (final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient()) { try { final ControllerDTO dto = apiClient.getController(); @@ -1182,7 +1198,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { logger.error("", e); } } - */ + */ authorizationIssue = e.getDescription(); } else if (e.getResponseCode() == FORBIDDEN_STATUS_CODE) { @@ -1192,7 +1208,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { logger.warn("{} When communicating with remote instance, got unexpected result. {}", new Object[]{this, e.getMessage()}); authorizationIssue = "Unable to determine Site-to-Site availability."; - } + } } } catch (final Exception e) { @@ -1321,7 +1337,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup { } private File getPeerPersistenceFile() { - final File stateDir = NiFiProperties.getInstance().getPersistentStateDirectory(); + final File stateDir = nifiProperties.getPersistentStateDirectory(); return new File(stateDir, getIdentifier() + ".peers"); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/groovy/org/apache/nifi/controller/StandardFlowSynchronizerSpec.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/groovy/org/apache/nifi/controller/StandardFlowSynchronizerSpec.groovy b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/groovy/org/apache/nifi/controller/StandardFlowSynchronizerSpec.groovy index 1ef622a..28b314c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/groovy/org/apache/nifi/controller/StandardFlowSynchronizerSpec.groovy +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/groovy/org/apache/nifi/controller/StandardFlowSynchronizerSpec.groovy @@ -31,9 +31,10 @@ import spock.lang.Specification import spock.lang.Unroll class StandardFlowSynchronizerSpec extends Specification { - + def setupSpec() { - System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties" + def propFile = StandardFlowSynchronizerSpec.class.getResource("/nifi.properties").getFile() + System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, propFile } def teardownSpec() { @@ -64,7 +65,8 @@ class StandardFlowSynchronizerSpec extends Specification { def Map<String, Connection> connectionMocksById = [:] def Map<String, List<Position>> bendPointPositionsByConnectionId = [:] // the unit under test - def flowSynchronizer = new StandardFlowSynchronizer(null) + def nifiProperties = NiFiProperties.createBasicNiFiProperties(null, null) + def flowSynchronizer = new StandardFlowSynchronizer(null,nifiProperties) when: "the flow is synchronized with the current state of the controller" flowSynchronizer.sync controller, proposedFlow, null 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/StandardFlowServiceTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java index 986fba4..79b59d7 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java @@ -74,7 +74,7 @@ public class StandardFlowServiceTest { @Before public void setup() throws Exception { - properties = NiFiProperties.getInstance(); + properties = NiFiProperties.createBasicNiFiProperties(null, null); variableRegistry = new FileBasedVariableRegistry(properties.getVariableRegistryPropertiesPaths()); mockFlowFileEventRepository = mock(FlowFileEventRepository.class); authorizer = mock(Authorizer.class); 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/TestFileSystemSwapManager.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFileSystemSwapManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFileSystemSwapManager.java index 15a8267..6c52def 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFileSystemSwapManager.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFileSystemSwapManager.java @@ -49,10 +49,9 @@ public class TestFileSystemSwapManager { @Test public void testBackwardCompatible() throws IOException { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); try (final InputStream fis = new FileInputStream(new File("src/test/resources/old-swap-file.swap")); - final DataInputStream in = new DataInputStream(new BufferedInputStream(fis))) { + final DataInputStream in = new DataInputStream(new BufferedInputStream(fis))) { final FlowFileQueue flowFileQueue = Mockito.mock(FlowFileQueue.class); Mockito.when(flowFileQueue.getIdentifier()).thenReturn("87bb99fe-412c-49f6-a441-d1b0af4e20b4"); @@ -91,7 +90,7 @@ public class TestFileSystemSwapManager { final SwapContents swappedIn; try (final FileInputStream fis = new FileInputStream(swapFile); - final DataInputStream dis = new DataInputStream(fis)) { + final DataInputStream dis = new DataInputStream(fis)) { swappedIn = FileSystemSwapManager.deserializeFlowFiles(dis, swapLocation, flowFileQueue, Mockito.mock(ResourceClaimManager.class)); } @@ -113,7 +112,6 @@ public class TestFileSystemSwapManager { } } - public class NopResourceClaimManager implements ResourceClaimManager { @Override @@ -162,8 +160,8 @@ public class TestFileSystemSwapManager { } } - private static class TestFlowFile implements FlowFileRecord { + private static final AtomicLong idGenerator = new AtomicLong(0L); private final long id = idGenerator.getAndIncrement(); @@ -172,13 +170,11 @@ public class TestFileSystemSwapManager { private final Map<String, String> attributes; private final long size; - public TestFlowFile(final Map<String, String> attributes, final long size) { this.attributes = attributes; this.size = size; } - @Override public long getId() { return id; 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/TestFlowController.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java index 71af93f..602ddce 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.controller; import org.apache.commons.io.IOUtils; @@ -46,7 +45,9 @@ import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.HashMap; import java.util.LinkedHashSet; +import java.util.Map; import java.util.Set; import org.apache.nifi.util.FileBasedVariableRegistry; @@ -65,21 +66,22 @@ public class TestFlowController { private FlowFileEventRepository flowFileEventRepo; private AuditService auditService; private StringEncryptor encryptor; - private NiFiProperties properties; + private NiFiProperties nifiProperties; private BulletinRepository bulletinRepo; private VariableRegistry variableRegistry; @Before public void setup() { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestFlowController.class.getResource("/nifi.properties").getFile()); flowFileEventRepo = Mockito.mock(FlowFileEventRepository.class); auditService = Mockito.mock(AuditService.class); - encryptor = StringEncryptor.createEncryptor(); - 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", ""); + final Map<String, String> otherProps = new HashMap<>(); + otherProps.put(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, MockProvenanceRepository.class.getName()); + otherProps.put("nifi.remote.input.socket.port", ""); + otherProps.put("nifi.remote.input.secure", ""); + nifiProperties = NiFiProperties.createBasicNiFiProperties(null, otherProps); + encryptor = StringEncryptor.createEncryptor(nifiProperties); User user1 = new User.Builder().identifier("user-id-1").identity("user-1").build(); User user2 = new User.Builder().identifier("user-id-2").identity("user-2").build(); @@ -118,12 +120,12 @@ public class TestFlowController { policies1.add(policy2); authorizer = new MockPolicyBasedAuthorizer(groups1, users1, policies1); - variableRegistry = new FileBasedVariableRegistry(properties.getVariableRegistryPropertiesPaths()); + variableRegistry = new FileBasedVariableRegistry(nifiProperties.getVariableRegistryPropertiesPaths()); bulletinRepo = Mockito.mock(BulletinRepository.class); - controller = FlowController.createStandaloneInstance(flowFileEventRepo, properties, authorizer, auditService, encryptor, bulletinRepo,variableRegistry); + controller = FlowController.createStandaloneInstance(flowFileEventRepo, nifiProperties, authorizer, auditService, encryptor, bulletinRepo, variableRegistry); - standardFlowSynchronizer = new StandardFlowSynchronizer(StringEncryptor.createEncryptor()); + standardFlowSynchronizer = new StandardFlowSynchronizer(StringEncryptor.createEncryptor(nifiProperties), nifiProperties); } @After @@ -271,7 +273,7 @@ public class TestFlowController { assertNotEquals(authFingerprint, authorizer.getFingerprint()); controller.shutdown(true); - controller = FlowController.createStandaloneInstance(flowFileEventRepo, properties, authorizer, auditService, encryptor, bulletinRepo,variableRegistry); + controller = FlowController.createStandaloneInstance(flowFileEventRepo, nifiProperties, authorizer, auditService, encryptor, bulletinRepo, variableRegistry); controller.synchronize(standardFlowSynchronizer, proposedDataFlow); assertEquals(authFingerprint, authorizer.getFingerprint()); } 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/TestStandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java index 3379f79..f6dc88e 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java @@ -41,6 +41,7 @@ import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.StandardProcessContext; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.util.MockPropertyValue; +import org.apache.nifi.util.NiFiProperties; import org.junit.Assert; import org.junit.Test; @@ -48,11 +49,11 @@ public class TestStandardProcessorNode { @Test(timeout = 10000) public void testStart() throws InterruptedException { - System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestStandardProcessorNode.class.getResource("/conf/nifi.properties").getFile()); final ProcessorThatThrowsExceptionOnScheduled processor = new ProcessorThatThrowsExceptionOnScheduled(); final String uuid = UUID.randomUUID().toString(); - final StandardProcessorNode procNode = new StandardProcessorNode(processor, uuid, createValidationContextFactory(), null, null); + final StandardProcessorNode procNode = new StandardProcessorNode(processor, uuid, createValidationContextFactory(), null, null, NiFiProperties.createBasicNiFiProperties(null, null)); final ScheduledExecutorService taskScheduler = new FlowEngine(2, "TestStandardProcessorNode", true); final StandardProcessContext processContext = new StandardProcessContext(procNode, null, null, null, null); 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/TestFileSystemRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java index 2a82aed..af33ccb 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java @@ -55,6 +55,8 @@ import ch.qos.logback.classic.Level; import ch.qos.logback.classic.Logger; import ch.qos.logback.classic.spi.ILoggingEvent; import ch.qos.logback.core.read.ListAppender; +import java.util.HashMap; +import java.util.Map; public class TestFileSystemRepository { @@ -65,14 +67,16 @@ public class TestFileSystemRepository { private FileSystemRepository repository = null; private StandardResourceClaimManager claimManager = null; private final File rootFile = new File("target/content_repository"); + private NiFiProperties nifiProperties; @Before public void setup() throws IOException { - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestFileSystemRepository.class.getResource("/conf/nifi.properties").getFile()); + nifiProperties = NiFiProperties.createBasicNiFiProperties(null, null); if (rootFile.exists()) { DiskUtils.deleteRecursively(rootFile); } - repository = new FileSystemRepository(); + repository = new FileSystemRepository(nifiProperties); claimManager = new StandardResourceClaimManager(); repository.initialize(claimManager); repository.purge(); @@ -94,49 +98,39 @@ public class TestFileSystemRepository { testAppender.setName("Test"); testAppender.start(); root.addAppender(testAppender); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, "1 millis"); + final NiFiProperties localProps = NiFiProperties.createBasicNiFiProperties(null, addProps); + repository = new FileSystemRepository(localProps); + repository.initialize(new StandardResourceClaimManager()); + repository.purge(); - final NiFiProperties properties = NiFiProperties.getInstance(); - final String originalCleanupFreq = properties.getProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); - properties.setProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, "1 millis"); - try { - repository = new FileSystemRepository(); - repository.initialize(new StandardResourceClaimManager()); - repository.purge(); - - - boolean messageFound = false; - String message = "The value of nifi.content.repository.archive.cleanup.frequency property " + boolean messageFound = false; + String message = "The value of nifi.content.repository.archive.cleanup.frequency property " + "is set to '1 millis' which is below the allowed minimum of 1 second (1000 milliseconds). " + "Minimum value of 1 sec will be used as scheduling interval for archive cleanup task."; - for (ILoggingEvent event : testAppender.list) { - String actualMessage = event.getFormattedMessage(); - if (actualMessage.equals(message)) { - assertEquals(event.getLevel(), Level.WARN); - messageFound = true; - break; - } - } - assertTrue(messageFound); - } finally { - if (originalCleanupFreq == null) { - properties.remove(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); - } else { - properties.setProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, originalCleanupFreq); + for (ILoggingEvent event : testAppender.list) { + String actualMessage = event.getFormattedMessage(); + if (actualMessage.equals(message)) { + assertEquals(event.getLevel(), Level.WARN); + messageFound = true; + break; } } + assertTrue(messageFound); } @Test public void testBogusFile() throws IOException { repository.shutdown(); - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestFileSystemRepository.class.getResource("/conf/nifi.properties").getFile()); File bogus = new File(rootFile, "bogus"); try { bogus.mkdir(); bogus.setReadable(false); - repository = new FileSystemRepository(); + repository = new FileSystemRepository(nifiProperties); repository.initialize(new StandardResourceClaimManager()); } finally { bogus.setReadable(true); @@ -198,7 +192,7 @@ public class TestFileSystemRepository { repository.shutdown(); Thread.sleep(1000L); - repository = new FileSystemRepository(); + repository = new FileSystemRepository(nifiProperties); repository.initialize(new StandardResourceClaimManager()); repository.purge(); @@ -206,7 +200,6 @@ public class TestFileSystemRepository { assertNotSame(claim1.getResourceClaim(), claim2.getResourceClaim()); } - @Test public void testWriteWithNoContent() throws IOException { final ContentClaim claim1 = repository.create(false); @@ -314,7 +307,6 @@ public class TestFileSystemRepository { assertTrue(Arrays.equals(expected, baos.toByteArray())); } - @Test public void testImportFromStream() throws IOException { final ContentClaim claim = repository.create(false); @@ -452,7 +444,7 @@ public class TestFileSystemRepository { // We are creating our own 'local' repository in this test so shut down the one created in the setup() method shutdown(); - repository = new FileSystemRepository() { + repository = new FileSystemRepository(nifiProperties) { @Override protected boolean archive(Path curPath) throws IOException { archivedPaths.add(curPath); @@ -494,7 +486,6 @@ public class TestFileSystemRepository { } } - @Test public void testWriteCannotProvideNullOutput() throws IOException { FileSystemRepository repository = null; @@ -504,7 +495,7 @@ public class TestFileSystemRepository { // We are creating our own 'local' repository in this test so shut down the one created in the setup() method shutdown(); - repository = new FileSystemRepository() { + repository = new FileSystemRepository(nifiProperties) { @Override protected boolean archive(Path curPath) throws IOException { if (getOpenStreamCount() > 0) { @@ -545,23 +536,33 @@ public class TestFileSystemRepository { } /** - * We have encountered a situation where the File System Repo is moving files to archive and then eventually - * aging them off while there is still an open file handle. This test is meant to replicate the conditions under + * We have encountered a situation where the File System Repo is moving + * files to archive and then eventually aging them off while there is still + * an open file handle. This test is meant to replicate the conditions under * which this would happen and verify that it is fixed. * - * The condition that caused this appears to be that a Process Session created a Content Claim and then did not write - * to it. It then decremented the claimant count (which reduced the count to 0). This was likely due to creating the - * claim in ProcessSession.write(FlowFile, StreamCallback) and then having an Exception thrown when the Process Session - * attempts to read the current Content Claim. In this case, it would not ever get to the point of calling - * FileSystemRepository.write(). + * The condition that caused this appears to be that a Process Session + * created a Content Claim and then did not write to it. It then decremented + * the claimant count (which reduced the count to 0). This was likely due to + * creating the claim in ProcessSession.write(FlowFile, StreamCallback) and + * then having an Exception thrown when the Process Session attempts to read + * the current Content Claim. In this case, it would not ever get to the + * point of calling FileSystemRepository.write(). * - * The above sequence of events is problematic because calling FileSystemRepository.create() will remove the Resource Claim - * from the 'writable claims queue' and expects that we will write to it. When we call FileSystemRepository.write() with that - * Resource Claim, we return an OutputStream that, when closed, will take care of adding the Resource Claim back to the - * 'writable claims queue' or otherwise close the FileOutputStream that is open for that Resource Claim. If FileSystemRepository.write() - * is never called, or if the OutputStream returned by that method is never closed, but the Content Claim is then decremented to 0, - * we can get into a situation where we do archive the content (because the claimant count is 0 and it is not in the 'writable claims queue') - * and then eventually age it off, without ever closing the OutputStream. We need to ensure that we do always close that Output Stream. + * The above sequence of events is problematic because calling + * FileSystemRepository.create() will remove the Resource Claim from the + * 'writable claims queue' and expects that we will write to it. When we + * call FileSystemRepository.write() with that Resource Claim, we return an + * OutputStream that, when closed, will take care of adding the Resource + * Claim back to the 'writable claims queue' or otherwise close the + * FileOutputStream that is open for that Resource Claim. If + * FileSystemRepository.write() is never called, or if the OutputStream + * returned by that method is never closed, but the Content Claim is then + * decremented to 0, we can get into a situation where we do archive the + * content (because the claimant count is 0 and it is not in the 'writable + * claims queue') and then eventually age it off, without ever closing the + * OutputStream. We need to ensure that we do always close that Output + * Stream. */ @Test public void testMarkDestructableDoesNotArchiveIfStreamOpenAndNotWrittenTo() throws IOException, InterruptedException { @@ -572,7 +573,7 @@ public class TestFileSystemRepository { // We are creating our own 'local' repository in this test so shut down the one created in the setup() method shutdown(); - repository = new FileSystemRepository() { + repository = new FileSystemRepository(nifiProperties) { @Override protected boolean archive(Path curPath) throws IOException { if (getOpenStreamCount() > 0) {
