[3/4] mesos git commit: Ensured executor adapter propagates error and shutdown messages.

[al...@apache.org]

Ensured executor adapter propagates error and shutdown messages.

Prior to this patch, if an error, kill, or shutdown occurred during
subscription / registration with the agent, it was not propagated back
to the executor if the v0_v1 executor adapter was used. This happened
because the adapter did not call the `connected` callback until after
successful registration and hence the executor did not even try to
send the `SUBSCRIBE` call, without which the adapter did not send any
events to the executor.

A fix is to call the `connected` callback if an error occurred or
shutdown / kill event arrived before the executor had subscribed.

Review: https://reviews.apache.org/r/64070/


Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/b2eddcfe
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/b2eddcfe
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/b2eddcfe

Branch: refs/heads/master
Commit: b2eddcfe0ede4725208ae33c8c7f56563ff10514
Parents: 47392cf
Author: Alexander Rukletsov <ru...@gmail.com>
Authored: Fri Dec 22 12:10:28 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Fri Dec 22 12:10:28 2017 +0100

----------------------------------------------------------------------
 src/executor/v0_v1executor.cpp | 41 ++++++++++++++++++++++++++++++++++++-
 1 file changed, 40 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/b2eddcfe/src/executor/v0_v1executor.cpp
----------------------------------------------------------------------
diff --git a/src/executor/v0_v1executor.cpp b/src/executor/v0_v1executor.cpp
index 61d5919..086cfc7 100644
--- a/src/executor/v0_v1executor.cpp
+++ b/src/executor/v0_v1executor.cpp
@@ -52,6 +52,7 @@ public:
       const function<void(const queue<Event>&)>& received)
     : ProcessBase(process::ID::generate("v0-to-v1-adapter")),
       callbacks {connected, disconnected, received},
+      connected(false),
       subscribeCall(false) {}
 
   virtual ~V0ToV1AdapterProcess() = default;
@@ -61,7 +62,10 @@ public:
       const mesos::FrameworkInfo& _frameworkInfo,
       const mesos::SlaveInfo& slaveInfo)
   {
-    callbacks.connected();
+    if (!connected) {
+      callbacks.connected();
+      connected = true;
+    }
 
     // We need these copies to populate the fields in `Event::Subscribed` upon
     // receiving a `reregistered()` callback later.
@@ -92,6 +96,7 @@ public:
     // disconnection from the agent.
     callbacks.disconnected();
     callbacks.connected();
+    connected = true;
 
     Event event;
     event.set_type(Event::SUBSCRIBED);
@@ -111,6 +116,17 @@ public:
 
   void killTask(const mesos::TaskID& taskId)
   {
+    // Logically an executor cannot receive any response from an agent if it
+    // is not connected. Since we have received `killTask`, we assume we are
+    // connected and trigger the `connected` callback to enable event delivery.
+    // This satisfies the invariant of the v1 interface that an executor can
+    // receive an event only after successfully connecting with the agent.
+    if (!connected) {
+      LOG(INFO) << "Implicitly connecting the executor to kill a task";
+      callbacks.connected();
+      connected = true;
+    }
+
     Event event;
     event.set_type(Event::KILL);
 
@@ -147,6 +163,17 @@ public:
 
   void shutdown()
   {
+    // Logically an executor cannot receive any response from an agent if it
+    // is not connected. Since we have received `shutdown`, we assume we are
+    // connected and trigger the `connected` callback to enable event delivery.
+    // This satisfies the invariant of the v1 interface that an executor can
+    // receive an event only after successfully connecting with the agent.
+    if (!connected) {
+      LOG(INFO) << "Implicitly connecting the executor to shut it down";
+      callbacks.connected();
+      connected = true;
+    }
+
     Event event;
     event.set_type(Event::SHUTDOWN);
 
@@ -155,6 +182,17 @@ public:
 
   void error(const string& message)
   {
+    // Logically an executor cannot receive any response from an agent if it
+    // is not connected. Since we have received `error`, we assume we are
+    // connected and trigger the `connected` callback to enable event delivery.
+    // This satisfies the invariant of the v1 interface that an executor can
+    // receive an event only after successfully connecting with the agent.
+    if (!connected) {
+      LOG(INFO) << "Implicitly connecting the executor to send an error";
+      callbacks.connected();
+      connected = true;
+    }
+
     Event event;
     event.set_type(Event::ERROR);
 
@@ -232,6 +270,7 @@ private:
   };
 
   Callbacks callbacks;
+  bool connected;
   bool subscribeCall;
   queue<Event> pending;
   Option<mesos::ExecutorInfo> executorInfo;