You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@asterixdb.apache.org by "abdullah alamoudi (Code Review)" <do...@asterixdb.incubator.apache.org> on 2018/07/15 20:32:50 UTC

Change in asterixdb[master]: [NO ISSUE][ING] Add timeout to stop active message

abdullah alamoudi has uploaded a new change for review.

  https://asterix-gerrit.ics.uci.edu/2762

Change subject: [NO ISSUE][ING] Add timeout to stop active message
......................................................................

[NO ISSUE][ING] Add timeout to stop active message

Change-Id: Ie0416d76670e945cb958f5a1c235201a3e016009
---
M asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
M asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveSourceOperatorNodePushable.java
M asterixdb/asterix-active/src/main/java/org/apache/asterix/active/IActiveRuntime.java
A asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/StopRuntimeContent.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedIntakeOperatorNodePushable.java
M hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
9 files changed, 100 insertions(+), 29 deletions(-)


  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/62/2762/1

diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
index 2aebc59..b44dfea 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
@@ -33,6 +33,7 @@
 import org.apache.asterix.active.message.ActiveManagerMessage;
 import org.apache.asterix.active.message.ActiveStatsRequestMessage;
 import org.apache.asterix.active.message.ActiveStatsResponse;
+import org.apache.asterix.active.message.StopRuntimeContent;
 import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.asterix.common.exceptions.RuntimeDataException;
 import org.apache.asterix.common.memory.ConcurrentFramePool;
@@ -139,7 +140,7 @@
         shutdown = true;
         runtimes.forEach((runtimeId, runtime) -> stopFutures.put(runtimeId, executor.submit(() -> {
             // we may already have been stopped- only stop once
-            stopIfRunning(runtimeId, runtime);
+            stopIfRunning(runtime, SHUTDOWN_TIMEOUT_SECS, TimeUnit.SECONDS);
             return null;
         })));
         stopFutures.entrySet().parallelStream().forEach(entry -> {
@@ -158,7 +159,8 @@
     }
 
     private void stopRuntime(ActiveManagerMessage message) {
-        ActiveRuntimeId runtimeId = (ActiveRuntimeId) message.getPayload();
+        StopRuntimeContent content = (StopRuntimeContent) message.getPayload();
+        ActiveRuntimeId runtimeId = content.getRuntimeId();
         IActiveRuntime runtime = runtimes.get(runtimeId);
         if (runtime == null) {
             LOGGER.warn("Request to stop runtime: " + runtimeId
@@ -167,21 +169,21 @@
         } else {
             executor.execute(() -> {
                 try {
-                    stopIfRunning(runtimeId, runtime);
-                } catch (Exception e) {
+                    stopIfRunning(runtime, content.getTimeout(), content.getUnit());
+                } catch (Throwable e) {
                     // TODO(till) Figure out a better way to handle failure to stop a runtime
-                    LOGGER.log(Level.WARN, "Failed to stop runtime: " + runtimeId, e);
+                    LOGGER.warn("Failed to stop runtime: {}", runtimeId, e);
                 }
             });
         }
     }
 
-    private void stopIfRunning(ActiveRuntimeId runtimeId, IActiveRuntime runtime)
+    private void stopIfRunning(IActiveRuntime runtime, long timeout, TimeUnit unit)
             throws HyracksDataException, InterruptedException {
-        if (runtimes.containsKey(runtimeId)) {
-            runtime.stop();
+        if (runtimes.containsKey(runtime.getRuntimeId())) {
+            runtime.stop(timeout, unit);
         } else {
-            LOGGER.info("Not stopping already stopped runtime " + runtimeId);
+            LOGGER.info("Not stopping already stopped runtime {}", runtime.getRuntimeId());
         }
     }
 
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveSourceOperatorNodePushable.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveSourceOperatorNodePushable.java
index 2f30df4..c3444ba 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveSourceOperatorNodePushable.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveSourceOperatorNodePushable.java
@@ -18,6 +18,8 @@
  */
 package org.apache.asterix.active;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.asterix.active.message.ActivePartitionMessage;
 import org.apache.asterix.active.message.ActivePartitionMessage.Event;
 import org.apache.asterix.common.api.INcApplicationContext;
@@ -62,10 +64,10 @@
     protected abstract void start() throws HyracksDataException, InterruptedException;
 
     @Override
-    public final void stop() throws HyracksDataException, InterruptedException {
+    public final void stop(long timeout, TimeUnit unit) throws HyracksDataException, InterruptedException {
         synchronized (this) {
             if (!done) {
-                abort();
+                abort(timeout, unit);
             }
             while (!done) {
                 wait();
@@ -75,11 +77,14 @@
 
     /**
      * called from a different thread. This method stops the active node and force the start() call to return
+     * 
+     * @param unit
+     * @param timeout
      *
      * @throws HyracksDataException
      * @throws InterruptedException
      */
-    protected abstract void abort() throws HyracksDataException, InterruptedException;
+    protected abstract void abort(long timeout, TimeUnit unit) throws HyracksDataException, InterruptedException;
 
     @Override
     public String toString() {
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/IActiveRuntime.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/IActiveRuntime.java
index f37b2e8..2da7193 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/IActiveRuntime.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/IActiveRuntime.java
@@ -18,6 +18,8 @@
  */
 package org.apache.asterix.active;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.job.JobId;
 
@@ -31,10 +33,15 @@
     /**
      * Stops the running activity
      *
+     * @param timeout
+     *            time for graceful stop. interrupt the runtime after that
+     * @param unit
+     *            unit of the timeout
+     *
      * @throws HyracksDataException
      * @throws InterruptedException
      */
-    void stop() throws HyracksDataException, InterruptedException;
+    void stop(long timeout, TimeUnit unit) throws HyracksDataException, InterruptedException;
 
     /**
      * @return the job id associated with this active runtime
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/StopRuntimeContent.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/StopRuntimeContent.java
new file mode 100644
index 0000000..891ce25
--- /dev/null
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/StopRuntimeContent.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.active.message;
+
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.asterix.active.ActiveRuntimeId;
+
+public class StopRuntimeContent implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+    private final ActiveRuntimeId runtimeId;
+    private final long timeout;
+    private final TimeUnit unit;
+
+    public StopRuntimeContent(ActiveRuntimeId runtimeId, long timeout, TimeUnit unit) {
+        this.runtimeId = runtimeId;
+        this.timeout = timeout;
+        this.unit = unit;
+    }
+
+    public ActiveRuntimeId getRuntimeId() {
+        return runtimeId;
+    }
+
+    public long getTimeout() {
+        return timeout;
+    }
+
+    public TimeUnit getUnit() {
+        return unit;
+    }
+}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index 66d3e81..e795830 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -40,6 +40,7 @@
 import org.apache.asterix.active.message.ActivePartitionMessage;
 import org.apache.asterix.active.message.ActivePartitionMessage.Event;
 import org.apache.asterix.active.message.ActiveStatsRequestMessage;
+import org.apache.asterix.active.message.StopRuntimeContent;
 import org.apache.asterix.common.cluster.IClusterStateManager;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.exceptions.ErrorCode;
@@ -73,7 +74,9 @@
     private static final EnumSet<ActivityState> TRANSITION_STATES = EnumSet.of(ActivityState.RESUMING,
             ActivityState.STARTING, ActivityState.STOPPING, ActivityState.RECOVERING, ActivityState.CANCELLING);
     private static final String DEFAULT_ACTIVE_STATS = "{\"Stats\":\"N/A\"}";
-    // finals
+    protected static final long STOP_MESSAGE_TIMEOUT = 2L;
+    protected static final long SUSPEND_MESSAGE_TIMEOUT = 5L;
+    protected static final TimeUnit TIMEOUT_UNIT = TimeUnit.MINUTES;
     protected final IClusterStateManager clusterStateManager;
     protected final ActiveNotificationHandler handler;
     protected final List<IActiveEntityEventSubscriber> subscribers = new ArrayList<>();
@@ -431,7 +434,8 @@
     protected abstract JobId compileAndStartJob(MetadataProvider metadataProvider) throws HyracksDataException;
 
     @SuppressWarnings("squid:S1181")
-    protected synchronized void doStop(MetadataProvider metadataProvider) throws HyracksDataException {
+    protected synchronized void doStop(MetadataProvider metadataProvider, long timeout, TimeUnit unit)
+            throws HyracksDataException {
         ActivityState intention = state;
         Set<ActivityState> waitFor;
         if (intention == ActivityState.STOPPING) {
@@ -444,7 +448,7 @@
         WaitForStateSubscriber subscriber = new WaitForStateSubscriber(this, waitFor);
         // Note: once we start sending stop messages, we can't go back until the entity is stopped
         try {
-            sendStopMessages(metadataProvider);
+            sendStopMessages(metadataProvider, timeout, unit);
             LOGGER.log(Level.DEBUG, "Waiting for its state to become " + waitFor);
             subscriber.sync();
             LOGGER.log(Level.DEBUG, "Disconnect has been completed " + waitFor);
@@ -465,7 +469,7 @@
         LOGGER.warn("Failure encountered while stopping {}", this, e);
     }
 
-    protected void sendStopMessages(MetadataProvider metadataProvider) throws Exception {
+    protected void sendStopMessages(MetadataProvider metadataProvider, long timeout, TimeUnit unit) throws Exception {
         ICcApplicationContext applicationCtx = metadataProvider.getApplicationContext();
         ICCMessageBroker messageBroker = (ICCMessageBroker) applicationCtx.getServiceContext().getMessageBroker();
         AlgebricksAbsolutePartitionConstraint runtimeLocations = getLocations();
@@ -473,9 +477,9 @@
         LOGGER.log(Level.INFO, "Sending stop messages to " + runtimeLocations);
         for (String location : runtimeLocations.getLocations()) {
             LOGGER.log(Level.INFO, "Sending to " + location);
-            messageBroker.sendApplicationMessageToNC(
-                    new ActiveManagerMessage(ActiveManagerMessage.Kind.STOP_ACTIVITY, getActiveRuntimeId(partition++)),
-                    location);
+            ActiveRuntimeId runtimeId = getActiveRuntimeId(partition++);
+            messageBroker.sendApplicationMessageToNC(new ActiveManagerMessage(ActiveManagerMessage.Kind.STOP_ACTIVITY,
+                    new StopRuntimeContent(runtimeId, timeout, unit)), location);
         }
     }
 
@@ -510,7 +514,7 @@
         } else if (state == ActivityState.RUNNING) {
             setState(ActivityState.STOPPING);
             try {
-                doStop(metadataProvider);
+                doStop(metadataProvider, STOP_MESSAGE_TIMEOUT, TIMEOUT_UNIT);
             } catch (Exception e) {
                 setState(ActivityState.STOPPED);
                 LOGGER.log(Level.ERROR, "Failed to stop the entity " + entityId, e);
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
index 47fc46f..88f1332 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
@@ -20,6 +20,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.asterix.active.ActivityState;
 import org.apache.asterix.active.EntityId;
@@ -62,7 +63,7 @@
     }
 
     @Override
-    protected void doStop(MetadataProvider metadataProvider) throws HyracksDataException {
+    protected void doStop(MetadataProvider metadataProvider, long timeout, TimeUnit unit) throws HyracksDataException {
         IActiveEntityEventSubscriber eventSubscriber =
                 new WaitForStateSubscriber(this, Collections.singleton(ActivityState.STOPPED));
         try {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
index c771a94..80806f3 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
@@ -21,6 +21,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.asterix.active.ActiveRuntimeId;
 import org.apache.asterix.active.ActivityState;
@@ -172,7 +173,7 @@
     }
 
     @Override
-    protected void sendStopMessages(MetadataProvider metadataProvider) throws Exception {
+    protected void sendStopMessages(MetadataProvider metadataProvider, long timeout, TimeUnit unit) throws Exception {
         step(onStop);
         failCompile(onStop);
         if (onStop == Behavior.RUNNING_JOB_FAIL) {
@@ -214,7 +215,7 @@
 
     @Override
     protected void doSuspend(MetadataProvider metadataProvider) throws HyracksDataException {
-        doStop(metadataProvider);
+        doStop(metadataProvider, SUSPEND_MESSAGE_TIMEOUT, TIMEOUT_UNIT);
     }
 
     @Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedIntakeOperatorNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedIntakeOperatorNodePushable.java
index 867fb60..98f75df 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedIntakeOperatorNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedIntakeOperatorNodePushable.java
@@ -18,6 +18,8 @@
  */
 package org.apache.asterix.external.operators;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.asterix.active.ActiveRuntimeId;
 import org.apache.asterix.active.ActiveSourceOperatorNodePushable;
 import org.apache.asterix.active.EntityId;
@@ -44,8 +46,6 @@
  */
 public class FeedIntakeOperatorNodePushable extends ActiveSourceOperatorNodePushable {
     private static final Logger LOGGER = LogManager.getLogger();
-    // TODO: Make configurable https://issues.apache.org/jira/browse/ASTERIXDB-2065
-    public static final int DEFAULT_ABORT_TIMEOUT = 60000;
     private final FeedIntakeOperatorDescriptor opDesc;
     private final FeedAdapter adapter;
     private boolean poisoned = false;
@@ -125,12 +125,12 @@
     }
 
     @Override
-    protected void abort() throws HyracksDataException, InterruptedException {
+    protected void abort(long timeout, TimeUnit unit) throws HyracksDataException, InterruptedException {
         LOGGER.info(runtimeId + " aborting...");
         synchronized (this) {
             poisoned = true;
             try {
-                if (!adapter.stop(DEFAULT_ABORT_TIMEOUT)) {
+                if (!adapter.stop(unit.toMillis(timeout))) {
                     LOGGER.info(runtimeId + " failed to stop adapter. interrupting the thread...");
                     taskThread.interrupt();
                 }
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
index c5a6de2..a16f4ef 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
@@ -65,7 +65,8 @@
                 writer.close();
             } catch (Throwable th) { // NOSONAR Will be suppressed
                 try {
-                    LOGGER.log(Level.WARN, "Failure closing a closeable resource", th);
+                    LOGGER.log(Level.WARN, "Failure closing a closeable resource of class {}",
+                            writer.getClass().getName(), th);
                 } catch (Throwable loggingFailure) { // NOSONAR: Ignore catching Throwable
                     // NOSONAR: Ignore logging failure
                 }

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/2762
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie0416d76670e945cb958f5a1c235201a3e016009
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: abdullah alamoudi <ba...@gmail.com>