Assume state sampler is always defined.

Author: tvalentyn

sdks/python/apache_beam/runners/worker/operations.py

@@ -445,19 +445,12 @@ def __init__(
     self.metrics_container = MetricsContainer(self.name_context.metrics_name())
 
     self.state_sampler = state_sampler
-    if self.state_sampler:
-      self.scoped_start_state = self.state_sampler.scoped_state(
-          self.name_context, 'start', metrics_container=self.metrics_container)
-      self.scoped_process_state = self.state_sampler.scoped_state(
-          self.name_context,
-          'process',
-          metrics_container=self.metrics_container)
-      self.scoped_finish_state = self.state_sampler.scoped_state(
-          self.name_context, 'finish', metrics_container=self.metrics_container)
-    else:
-      self.scoped_start_state = statesampler.NOOP_SCOPED_STATE
-      self.scoped_process_state = statesampler.NOOP_SCOPED_STATE
-      self.scoped_finish_state = statesampler.NOOP_SCOPED_STATE
+    self.scoped_start_state = self.state_sampler.scoped_state(
+        self.name_context, 'start', metrics_container=self.metrics_container)
+    self.scoped_process_state = self.state_sampler.scoped_state(
+        self.name_context, 'process', metrics_container=self.metrics_container)
+    self.scoped_finish_state = self.state_sampler.scoped_state(
+        self.name_context, 'finish', metrics_container=self.metrics_container)
     # TODO(ccy): the '-abort' state can be added when the abort is supported in
     # Operations.
     self.receivers = []  # type: List[ConsumerSet]