Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3232#discussion_r98630616
--- Diff:
flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py
---
@@ -27,17 +27,48 @@
import sys
from struct import pack
+
+class EnvironmentContainer(object):
+ """Keeps track of which ExecutionEnvironment is being run."""
+
+ environment_counter = 0
+ environment_id_to_execute = None
+
+ def create_environment(self):
+ env = Environment(self, self.environment_counter)
+ self.environment_counter += 1
+ return env
+
+ def is_executing(self):
+ """Checks if we are waiting for a certain environment to be
executed."""
+ return not self.environment_id_to_execute is None
+
+ def fetch_next_environment(self, calling_environment_id):
+ """Checks (and if necessary, fetches) the next environment to be
executed."""
+ if not self.is_executing():
+ self.environment_id_to_execute =
int(sys.stdin.readline().rstrip('\n'))
+
+ if self.environment_id_to_execute == calling_environment_id:
+ self.environment_id_to_execute = None
+ return True
+
+ return False
+
+
+container = EnvironmentContainer()
+
def get_environment():
"""
Creates an execution environment that represents the context in which
the program is currently executed.
-
+
:return:The execution environment of the context in which the program
is executed.
"""
- return Environment()
+ global container
--- End diff --
doesn't have to be global since we don't assign to `container` within
`get_environment()`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
