[
https://issues.apache.org/jira/browse/NIFI-1164?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15072828#comment-15072828
]
ASF GitHub Bot commented on NIFI-1164:
--------------------------------------
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/141#discussion_r48486134
--- Diff:
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
---
@@ -229,7 +244,106 @@ public ControllerServiceState getState() {
}
@Override
- public void setState(final ControllerServiceState state) {
- this.stateRef.set(state);
+ public boolean isActive() {
+ return this.active.get();
+ }
+
+ /**
+ * Will atomically enable this service by invoking its @OnEnabled
operation.
+ * It uses CAS operation on {@link #stateRef} to transition this
service
+ * from DISABLED to ENABLING state. If such transition succeeds the
service
+ * will be marked as 'active' (see {@link
ControllerServiceNode#isActive()}).
+ * If such transition doesn't succeed then no enabling logic will be
+ * performed and the method will exit. In other words it is safe to
invoke
+ * this operation multiple times and from multiple threads.
+ * <br>
+ * This operation will also perform re-try of service enabling in the
event
+ * of exception being thrown by previous invocation of @OnEnabled.
+ * <br>
+ * Upon successful invocation of @OnEnabled this service will be
transitioned to
+ * ENABLED state.
+ * <br>
+ * In the event where enabling took longer then expected by the user
and such user
+ * initiated disable operation, this service will be automatically
disabled as soon
+ * as it reached ENABLED state.
+ */
+ @Override
+ public void enable(final ScheduledExecutorService scheduler, final
long administrativeYieldMillis, final Heartbeater heartbeater) {
+ if (this.stateRef.compareAndSet(ControllerServiceState.DISABLED,
ControllerServiceState.ENABLING)){
+ this.active.set(true);
+ final ConfigurationContext configContext = new
StandardConfigurationContext(this, this.serviceProvider, null);
+ scheduler.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+
ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class,
getControllerServiceImplementation(), configContext);
+ if
(stateRef.compareAndSet(ControllerServiceState.ENABLING,
ControllerServiceState.ENABLED)) {
+ heartbeater.heartbeat();
+ } else {
+ LOG.debug("Disabling service " + this + "
after it has been enabled due to disable action being initiated.");
+ // Can only happen if user initiated DISABLE
operation before service finished enabling. It's state will be
+ // set to DISABLING (see disable() operation)
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnDisabled.class,
getControllerServiceImplementation(), configContext);
+ stateRef.set(ControllerServiceState.DISABLED);
+ }
+ } catch (Exception e) {
+ final Throwable cause = e instanceof
InvocationTargetException ? e.getCause() : e;
+ final ComponentLog componentLog = new
SimpleProcessLogger(getIdentifier(), StandardControllerServiceNode.this);
+ componentLog.error("Failed to invoke @OnEnabled
method due to {}", cause);
+ LOG.error("Failed to invoke @OnEnabled method of
{} due to {}", getControllerServiceImplementation(), cause.toString());
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnDisabled.class,
getControllerServiceImplementation(), configContext);
+ if (isActive()) {
+ scheduler.schedule(this,
administrativeYieldMillis, TimeUnit.MILLISECONDS);
+ }
+ else {
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnDisabled.class,
getControllerServiceImplementation(), configContext);
+ stateRef.set(ControllerServiceState.DISABLED);
+ }
+ }
+ }
+ });
+ }
+ }
+
+ /**
+ * Will atomically disable this service by invoking its @OnDisabled
operation.
+ * It uses CAS operation on {@link #stateRef} to transition this
service
+ * from ENABLED to DISABLING state. If such transition succeeds the
service
+ * will be de-activated (see {@link ControllerServiceNode#isActive()}).
+ * If such transition doesn't succeed (the service is still in
ENABLING state)
+ * then the service will still be transitioned to DISABLING state to
ensure that
+ * no other transition could happen on this service. However in such
event
+ * (e.g., its @OnEnabled finally succeeded), the {@link
#enable(ScheduledExecutorService, long, Heartbeater)}
+ * operation will initiate service disabling javadoc for (see {@link
#enable(ScheduledExecutorService, long, Heartbeater)}
+ * <br>
+ * Upon successful invocation of @OnDisabled this service will be
transitioned to
+ * DISABLED state.
+ */
+ @Override
+ public void disable(final ScheduledExecutorService scheduler, final
Heartbeater heartbeater) {
+ this.active.set(false); // de-activating regardless of CAS
operation
+ // that follows since this operation will
always result in service state being DISABLING
+ if (this.stateRef.compareAndSet(ControllerServiceState.ENABLED,
ControllerServiceState.DISABLING)) {
+ final ConfigurationContext configContext = new
StandardConfigurationContext(this, this.serviceProvider, null);
+ scheduler.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+
ReflectionUtils.invokeMethodsWithAnnotation(OnDisabled.class,
StandardControllerServiceNode.this.getControllerServiceImplementation(),
configContext);
+ } catch (Exception e) {
+ final Throwable cause = e instanceof
InvocationTargetException ? e.getCause() : e;
+ final ComponentLog componentLog = new
SimpleProcessLogger(getIdentifier(), StandardControllerServiceNode.this);
+ componentLog.error("Failed to invoke @OnDisabled
method due to {}", cause);
+ LOG.error("Failed to invoke @OnDisabled method of
{} due to {}", getControllerServiceImplementation(), cause.toString());
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnDisabled.class,
getControllerServiceImplementation(), configContext);
+ } finally {
+ stateRef.set(ControllerServiceState.DISABLED);
+ heartbeater.heartbeat();
+ }
+ }
+ });
+ } else {
+ this.stateRef.compareAndSet(ControllerServiceState.ENABLING,
ControllerServiceState.DISABLING);
--- End diff --
@olegz I think there was a case that was missed here. Assume that we have
two threads: E at line 280 and D and line 326. In this case, if D executes it
will return false from compareAndSet. Then, if thread E executes, it will
change the state from ENABLING to ENABLED. Thread D then executes line 346 and
gets a value of false. At this point, Thread D returns from the disable()
method, but it never did disable the service. This appears to be a race
condition that we need to tackle.
> Race condition during initialization of ControllerServices - (part due)
> -----------------------------------------------------------------------
>
> Key: NIFI-1164
> URL: https://issues.apache.org/jira/browse/NIFI-1164
> Project: Apache NiFi
> Issue Type: Bug
> Components: Core Framework
> Affects Versions: 0.3.0
> Reporter: Oleg Zhurakousky
> Assignee: Oleg Zhurakousky
> Fix For: 0.5.0
>
>
> Similar to NIFI-1143, but the exception is happening in the different place
> {code}testConcurrencyWithEnablingReferencingServicesGraph(org.apache.nifi.controller.service.TestStandardControllerServiceProvider)
> Time elapsed: 0.162 sec <<< ERROR!
> java.lang.IllegalStateException: ServiceB[id=4] cannot be enabled because it
> is not disabled
> at
> org.apache.nifi.controller.service.StandardControllerServiceNode.verifyCanEnable(StandardControllerServiceNode.java:183)
> at
> org.apache.nifi.controller.service.StandardControllerServiceProvider.enableReferencingServices(StandardControllerServiceProvider.java:549)
> at
> org.apache.nifi.controller.service.StandardControllerServiceProvider.enableReferencingServices(StandardControllerServiceProvider.java:544)
> at
> org.apache.nifi.controller.service.TestStandardControllerServiceProvider.testEnableReferencingServicesGraph(TestStandardControllerServiceProvider.java:154)
> at
> org.apache.nifi.controller.service.TestStandardControllerServiceProvider.testConcurrencyWithEnablingReferencingServicesGraph(TestStandardControllerServiceProvider.java:124)
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
