Add review comments

sdks/python/apache_beam/runners/direct/clock.py

@@ -25,25 +25,27 @@
 
 
 class Clock(object):
-  def current_time(self):
+  def time(self):
+    """Returns the number of milliseconds since epoch."""
     raise NotImplementedError()
 
-  def advance_time(self):
+  def advance_time(self, advance_by):
+    """Advances the clock by a number of miliseconds."""
     raise NotImplementedError()
 
 
 class RealClock(object):
-  def current_time(self):
-    return time.time()
+  def time(self):
+    return int(time.time() * 1000)
 
 
 class TestClock(object):
   """Clock used for Testing"""
-  def __init__(self, current=0):
-    self._current = current
+  def __init__(self, current_time=0):
+    self._current_time = current_time
 
-  def current_time(self):
-    return self._current
+  def time(self):
+    return self._current_time
 
   def advance_time(self, advance_by):
-    self._current += advance_by
+    self._current_time += advance_by

sdks/python/apache_beam/runners/direct/watermark_manager.py

@@ -249,7 +249,7 @@ def refresh(self):
 
   @property
   def synchronized_processing_output_time(self):
-    return self._clock.current_time()
+    return self._clock.time()
 
   def extract_transform_timers(self):
     """Extracts fired timers and reports of any timers set per transform."""
@@ -259,7 +259,7 @@ def extract_transform_timers(self):
       for encoded_key, state in self._keyed_states.iteritems():
         timers, had_realtime_timer = state.get_timers(
             watermark=self._input_watermark,
-            current_time=self._clock.current_time())
+            current_time=self._clock.time())
         if had_realtime_timer:
           has_realtime_timer = True
         for expired in timers: