Fix error for Regex test

Author: ktalluri456

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

@@ -83,7 +83,6 @@ cdef class Operation(object):
   cdef readonly object scoped_start_state
   cdef readonly object scoped_process_state
   cdef readonly object scoped_finish_state
-  cdef readonly object scoped_timer_processing_state
 
   cdef readonly object data_sampler
 
@@ -118,6 +117,7 @@ cdef class DoOperation(Operation):
   cdef dict timer_specs
   cdef public object input_info
   cdef object fn
+  cdef readonly object scoped_timer_processing_state
 
 
 cdef class SdfProcessSizedElements(DoOperation):

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

@@ -49,6 +49,7 @@
 from apache_beam.runners.worker import opcounters
 from apache_beam.runners.worker import operation_specs
 from apache_beam.runners.worker import sideinputs
+from apache_beam.runners.worker import statesampler
 from apache_beam.runners.worker.data_sampler import DataSampler
 from apache_beam.transforms import sideinputs as apache_sideinputs
 from apache_beam.transforms import combiners
@@ -451,9 +452,15 @@ def __init__(
       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.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
     # TODO(ccy): the '-abort' state can be added when the abort is supported in
     # Operations.
     self.receivers = []  # type: List[ConsumerSet]
@@ -812,9 +819,9 @@ def __init__(
     self.user_state_context = user_state_context
     self.tagged_receivers = None  # type: Optional[_TaggedReceivers]
     # A mapping of timer tags to the input "PCollections" they come in on.
-    # Force clean rebuild 
+    # Force clean rebuild
     self.input_info = None  # type: Optional[OpInputInfo]
-    self.scoped_timer_processing_state = None
+    self.scoped_timer_processing_state = statesampler.NOOP_SCOPED_STATE
     if self.state_sampler:
       self.scoped_timer_processing_state = self.state_sampler.scoped_state(
           self.name_context,

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

@@ -135,8 +135,7 @@ def scoped_state(
       state_name: str,
       io_target=None,
       metrics_container: Optional['MetricsContainer'] = None,
-      suffix: str = '-msecs'
-  ) -> statesampler_impl.ScopedState:
+      suffix: str = '-msecs') -> statesampler_impl.ScopedState:
     """Returns a ScopedState object associated to a Step and a State.
 
     Args:
@@ -171,3 +170,17 @@ def commit_counters(self) -> None:
     for state in self._states_by_name.values():
       state_msecs = int(1e-6 * state.nsecs)
       state.counter.update(state_msecs - state.counter.value())
+
+
+class NoOpScopedState:
+  def __enter__(self):
+    pass
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    pass
+
+  def sampled_msecs_int(self):
+    return 0
+
+
+NOOP_SCOPED_STATE = NoOpScopedState()

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

@@ -139,8 +139,7 @@ def test_timer_sampler(self):
     margin_of_error = 0.25
 
     sampler.start()
-    with sampler.scoped_state(
-        'step1', 'process-timers', suffix='-millis'):
+    with sampler.scoped_state('step1', 'process-timers', suffix='-millis'):
       time.sleep(state_duration_ms / 1000)
     sampler.stop()
     sampler.commit_counters()
@@ -151,8 +150,8 @@ def test_timer_sampler(self):
 
     # Test that sampled state timings are close to their expected values.
     expected_counter_values = {
-        CounterName('process-timers-millis', step_name='step1', stage_name='timer'):
-            state_duration_ms,
+        CounterName(
+            'process-timers-millis', step_name='step1', stage_name='timer'): state_duration_ms,
     }
     for counter in counter_factory.get_counters():
       self.assertIn(counter.name, expected_counter_values)
@@ -164,7 +163,6 @@ def test_timer_sampler(self):
       self.assertLess(actual_value, expected_value * (1.0 + margin_of_error))
 
 
-
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)
   unittest.main()