beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/2] beam git commit: [BEAM-2997] Encapsulate enums within a message so that C++/Python have meaningful namespaces when importing.
Date Fri, 29 Sep 2017 17:15:27 GMT
Repository: beam
Updated Branches:
  refs/heads/master 294f6339e -> 25b9c35a9


[BEAM-2997] Encapsulate enums within a message so that C++/Python have meaningful namespaces
when importing.


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

Branch: refs/heads/master
Commit: 6fddd4ed951c6e5618add72f49211bedcf046edc
Parents: da531b7
Author: Luke Cwik <lcwik@google.com>
Authored: Wed Sep 27 13:33:16 2017 -0700
Committer: Luke Cwik <lcwik@google.com>
Committed: Thu Sep 28 11:40:43 2017 -0700

----------------------------------------------------------------------
 .../construction/DisplayDataTranslation.java    |   2 +-
 .../construction/PCollectionTranslation.java    |   8 +-
 .../core/construction/ParDoTranslation.java     |  12 +-
 .../core/construction/ReadTranslation.java      |   8 +-
 .../core/construction/TriggerTranslation.java   |   8 +-
 .../WindowingStrategyTranslation.java           |  34 ++--
 .../core/construction/ReadTranslationTest.java  |   4 +-
 .../fn-api/src/main/proto/beam_fn_api.proto     |  42 ++--
 .../src/main/proto/beam_job_api.proto           |  35 ++--
 .../src/main/proto/beam_runner_api.proto        | 200 ++++++++++---------
 .../fn/harness/logging/BeamFnLoggingClient.java |  16 +-
 .../logging/BeamFnLoggingClientTest.java        |   4 +-
 sdks/python/apache_beam/io/iobase.py            |   4 +-
 sdks/python/apache_beam/pvalue.py               |   2 +-
 .../python_rpc_direct_runner.py                 |   2 +-
 .../experimental/python_rpc_direct/server.py    |   2 +-
 .../portability/universal_local_runner.py       |   6 +-
 .../apache_beam/runners/worker/log_handler.py   |  10 +-
 .../runners/worker/log_handler_test.py          |   3 +-
 sdks/python/apache_beam/transforms/core.py      |   6 +-
 sdks/python/apache_beam/transforms/trigger.py   |   4 +-
 sdks/python/apache_beam/transforms/window.py    |   6 +-
 22 files changed, 219 insertions(+), 199 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java
index ff7f9f2..5186caf 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java
@@ -32,7 +32,7 @@ public class DisplayDataTranslation {
             RunnerApi.DisplayData.Item.newBuilder()
                 .setId(RunnerApi.DisplayData.Identifier.newBuilder().setKey("stubImplementation"))
                 .setLabel("Stub implementation")
-                .setType(RunnerApi.DisplayData.Type.BOOLEAN)
+                .setType(RunnerApi.DisplayData.Type.Enum.BOOLEAN)
                 .setValue(Any.pack(BoolValue.newBuilder().setValue(true).build())))
         .build();
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
index c256e4c..84b3386 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
@@ -62,19 +62,19 @@ public class PCollectionTranslation {
     return fromProto(pCollection.getIsBounded());
   }
 
-  static RunnerApi.IsBounded toProto(IsBounded bounded) {
+  static RunnerApi.IsBounded.Enum toProto(IsBounded bounded) {
     switch (bounded) {
       case BOUNDED:
-        return RunnerApi.IsBounded.BOUNDED;
+        return RunnerApi.IsBounded.Enum.BOUNDED;
       case UNBOUNDED:
-        return RunnerApi.IsBounded.UNBOUNDED;
+        return RunnerApi.IsBounded.Enum.UNBOUNDED;
       default:
         throw new IllegalArgumentException(
             String.format("Unknown %s %s", IsBounded.class.getSimpleName(), bounded));
     }
   }
 
-  static IsBounded fromProto(RunnerApi.IsBounded isBounded) {
+  static IsBounded fromProto(RunnerApi.IsBounded.Enum isBounded) {
     switch (isBounded) {
       case BOUNDED:
         return IsBounded.BOUNDED;

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
index 037ffe3..714c59d 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
@@ -430,14 +430,14 @@ public class ParDoTranslation {
     return RunnerApi.TimerSpec.newBuilder().setTimeDomain(toProto(timer.getTimeDomain())).build();
   }
 
-  private static RunnerApi.TimeDomain toProto(TimeDomain timeDomain) {
+  private static RunnerApi.TimeDomain.Enum toProto(TimeDomain timeDomain) {
     switch(timeDomain) {
       case EVENT_TIME:
-        return RunnerApi.TimeDomain.EVENT_TIME;
+        return RunnerApi.TimeDomain.Enum.EVENT_TIME;
       case PROCESSING_TIME:
-        return RunnerApi.TimeDomain.PROCESSING_TIME;
+        return RunnerApi.TimeDomain.Enum.PROCESSING_TIME;
       case SYNCHRONIZED_PROCESSING_TIME:
-        return RunnerApi.TimeDomain.SYNCHRONIZED_PROCESSING_TIME;
+        return RunnerApi.TimeDomain.Enum.SYNCHRONIZED_PROCESSING_TIME;
       default:
         throw new IllegalArgumentException("Unknown time domain");
     }
@@ -486,13 +486,13 @@ public class ParDoTranslation {
         new Cases.WithDefault<Optional<RunnerApi.Parameter>>() {
           @Override
           public Optional<RunnerApi.Parameter> dispatch(WindowParameter p) {
-            return Optional.of(RunnerApi.Parameter.newBuilder().setType(Type.WINDOW).build());
+            return Optional.of(RunnerApi.Parameter.newBuilder().setType(Type.Enum.WINDOW).build());
           }
 
           @Override
           public Optional<RunnerApi.Parameter> dispatch(RestrictionTrackerParameter
p) {
             return Optional.of(
-                RunnerApi.Parameter.newBuilder().setType(Type.RESTRICTION_TRACKER).build());
+                RunnerApi.Parameter.newBuilder().setType(Type.Enum.RESTRICTION_TRACKER).build());
           }
 
           @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
index 06d1074..4cc31e8 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
@@ -53,14 +53,14 @@ public class ReadTranslation {
 
   public static ReadPayload toProto(Read.Bounded<?> read) {
     return ReadPayload.newBuilder()
-        .setIsBounded(IsBounded.BOUNDED)
+        .setIsBounded(IsBounded.Enum.BOUNDED)
         .setSource(toProto(read.getSource()))
         .build();
   }
 
   public static ReadPayload toProto(Read.Unbounded<?> read) {
     return ReadPayload.newBuilder()
-        .setIsBounded(IsBounded.UNBOUNDED)
+        .setIsBounded(IsBounded.Enum.UNBOUNDED)
         .setSource(toProto(read.getSource()))
         .build();
   }
@@ -88,7 +88,7 @@ public class ReadTranslation {
 
   public static BoundedSource<?> boundedSourceFromProto(ReadPayload payload)
       throws InvalidProtocolBufferException {
-    checkArgument(payload.getIsBounded().equals(IsBounded.BOUNDED));
+    checkArgument(payload.getIsBounded().equals(IsBounded.Enum.BOUNDED));
     return (BoundedSource<?>) SerializableUtils.deserializeFromByteArray(
         payload
             .getSource()
@@ -135,7 +135,7 @@ public class ReadTranslation {
 
   public static UnboundedSource<?, ?> unboundedSourceFromProto(ReadPayload payload)
       throws InvalidProtocolBufferException {
-    checkArgument(payload.getIsBounded().equals(IsBounded.UNBOUNDED));
+    checkArgument(payload.getIsBounded().equals(IsBounded.Enum.UNBOUNDED));
     return (UnboundedSource<?, ?>) SerializableUtils.deserializeFromByteArray(
         payload
             .getSource()

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/TriggerTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/TriggerTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/TriggerTranslation.java
index 777b165..b23f686 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/TriggerTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/TriggerTranslation.java
@@ -123,14 +123,14 @@ public class TriggerTranslation implements Serializable {
           .build();
     }
 
-    private RunnerApi.TimeDomain convertTimeDomain(TimeDomain timeDomain) {
+    private RunnerApi.TimeDomain.Enum convertTimeDomain(TimeDomain timeDomain) {
       switch (timeDomain) {
         case EVENT_TIME:
-          return RunnerApi.TimeDomain.EVENT_TIME;
+          return RunnerApi.TimeDomain.Enum.EVENT_TIME;
         case PROCESSING_TIME:
-          return RunnerApi.TimeDomain.PROCESSING_TIME;
+          return RunnerApi.TimeDomain.Enum.PROCESSING_TIME;
         case SYNCHRONIZED_PROCESSING_TIME:
-          return RunnerApi.TimeDomain.SYNCHRONIZED_PROCESSING_TIME;
+          return RunnerApi.TimeDomain.Enum.SYNCHRONIZED_PROCESSING_TIME;
         default:
           throw new IllegalArgumentException(String.format("Unknown time domain: %s", timeDomain));
       }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
index 7e02da8..1b4786c 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
@@ -51,7 +51,7 @@ import org.joda.time.Duration;
 /** Utilities for working with {@link WindowingStrategy WindowingStrategies}. */
 public class WindowingStrategyTranslation implements Serializable {
 
-  public static AccumulationMode fromProto(RunnerApi.AccumulationMode proto) {
+  public static AccumulationMode fromProto(RunnerApi.AccumulationMode.Enum proto) {
     switch (proto) {
       case DISCARDING:
         return AccumulationMode.DISCARDING_FIRED_PANES;
@@ -71,12 +71,12 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static RunnerApi.AccumulationMode toProto(AccumulationMode accumulationMode) {
+  public static RunnerApi.AccumulationMode.Enum toProto(AccumulationMode accumulationMode)
{
     switch (accumulationMode) {
       case DISCARDING_FIRED_PANES:
-        return RunnerApi.AccumulationMode.DISCARDING;
+        return RunnerApi.AccumulationMode.Enum.DISCARDING;
       case ACCUMULATING_FIRED_PANES:
-        return RunnerApi.AccumulationMode.ACCUMULATING;
+        return RunnerApi.AccumulationMode.Enum.ACCUMULATING;
       default:
         throw new IllegalArgumentException(
             String.format(
@@ -87,12 +87,12 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static RunnerApi.ClosingBehavior toProto(ClosingBehavior closingBehavior) {
+  public static RunnerApi.ClosingBehavior.Enum toProto(ClosingBehavior closingBehavior) {
     switch (closingBehavior) {
       case FIRE_ALWAYS:
-        return RunnerApi.ClosingBehavior.EMIT_ALWAYS;
+        return RunnerApi.ClosingBehavior.Enum.EMIT_ALWAYS;
       case FIRE_IF_NON_EMPTY:
-        return RunnerApi.ClosingBehavior.EMIT_IF_NONEMPTY;
+        return RunnerApi.ClosingBehavior.Enum.EMIT_IF_NONEMPTY;
       default:
         throw new IllegalArgumentException(
             String.format(
@@ -103,7 +103,7 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static ClosingBehavior fromProto(RunnerApi.ClosingBehavior proto) {
+  public static ClosingBehavior fromProto(RunnerApi.ClosingBehavior.Enum proto) {
     switch (proto) {
       case EMIT_ALWAYS:
         return ClosingBehavior.FIRE_ALWAYS;
@@ -123,12 +123,12 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static RunnerApi.OnTimeBehavior toProto(OnTimeBehavior onTimeBehavior) {
+  public static RunnerApi.OnTimeBehavior.Enum toProto(OnTimeBehavior onTimeBehavior) {
     switch (onTimeBehavior) {
       case FIRE_ALWAYS:
-        return RunnerApi.OnTimeBehavior.FIRE_ALWAYS;
+        return RunnerApi.OnTimeBehavior.Enum.FIRE_ALWAYS;
       case FIRE_IF_NON_EMPTY:
-        return RunnerApi.OnTimeBehavior.FIRE_IF_NONEMPTY;
+        return RunnerApi.OnTimeBehavior.Enum.FIRE_IF_NONEMPTY;
       default:
         throw new IllegalArgumentException(
             String.format(
@@ -139,7 +139,7 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static OnTimeBehavior fromProto(RunnerApi.OnTimeBehavior proto) {
+  public static OnTimeBehavior fromProto(RunnerApi.OnTimeBehavior.Enum proto) {
     switch (proto) {
       case FIRE_ALWAYS:
         return OnTimeBehavior.FIRE_ALWAYS;
@@ -159,14 +159,14 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static RunnerApi.OutputTime toProto(TimestampCombiner timestampCombiner) {
+  public static RunnerApi.OutputTime.Enum toProto(TimestampCombiner timestampCombiner) {
     switch(timestampCombiner) {
       case EARLIEST:
-        return OutputTime.EARLIEST_IN_PANE;
+        return OutputTime.Enum.EARLIEST_IN_PANE;
       case END_OF_WINDOW:
-        return OutputTime.END_OF_WINDOW;
+        return OutputTime.Enum.END_OF_WINDOW;
       case LATEST:
-        return OutputTime.LATEST_IN_PANE;
+        return OutputTime.Enum.LATEST_IN_PANE;
       default:
         throw new IllegalArgumentException(
             String.format(
@@ -176,7 +176,7 @@ public class WindowingStrategyTranslation implements Serializable {
     }
   }
 
-  public static TimestampCombiner timestampCombinerFromProto(RunnerApi.OutputTime proto)
{
+  public static TimestampCombiner timestampCombinerFromProto(RunnerApi.OutputTime.Enum proto)
{
     switch (proto) {
       case EARLIEST_IN_PANE:
         return TimestampCombiner.EARLIEST;

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ReadTranslationTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ReadTranslationTest.java
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ReadTranslationTest.java
index 3eee78c..22c79b3 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ReadTranslationTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ReadTranslationTest.java
@@ -75,7 +75,7 @@ public class ReadTranslationTest {
     BoundedSource<?> boundedSource = (BoundedSource<?>) this.source;
     Read.Bounded<?> boundedRead = Read.from(boundedSource);
     ReadPayload payload = ReadTranslation.toProto(boundedRead);
-    assertThat(payload.getIsBounded(), equalTo(RunnerApi.IsBounded.BOUNDED));
+    assertThat(payload.getIsBounded(), equalTo(RunnerApi.IsBounded.Enum.BOUNDED));
     BoundedSource<?> deserializedSource = ReadTranslation.boundedSourceFromProto(payload);
     assertThat(deserializedSource, Matchers.<Source<?>>equalTo(source));
   }
@@ -86,7 +86,7 @@ public class ReadTranslationTest {
     UnboundedSource<?, ?> unboundedSource = (UnboundedSource<?, ?>) this.source;
     Read.Unbounded<?> unboundedRead = Read.from(unboundedSource);
     ReadPayload payload = ReadTranslation.toProto(unboundedRead);
-    assertThat(payload.getIsBounded(), equalTo(RunnerApi.IsBounded.UNBOUNDED));
+    assertThat(payload.getIsBounded(), equalTo(RunnerApi.IsBounded.Enum.UNBOUNDED));
     UnboundedSource<?, ?> deserializedSource = ReadTranslation.unboundedSourceFromProto(payload);
     assertThat(deserializedSource, Matchers.<Source<?>>equalTo(source));
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/common/fn-api/src/main/proto/beam_fn_api.proto
----------------------------------------------------------------------
diff --git a/sdks/common/fn-api/src/main/proto/beam_fn_api.proto b/sdks/common/fn-api/src/main/proto/beam_fn_api.proto
index f2bbd3c..9d4c5f6 100644
--- a/sdks/common/fn-api/src/main/proto/beam_fn_api.proto
+++ b/sdks/common/fn-api/src/main/proto/beam_fn_api.proto
@@ -630,29 +630,31 @@ message LogEntry {
   // common set of "good enough" severity levels so that logging front ends
   // can provide filtering and searching across log types. Users of the API are
   // free not to use all severity levels in their log messages.
-  enum Severity {
-    SEVERITY_UNSPECIFIED = 0;
-    // Trace level information, also the default log level unless
-    // another severity is specified.
-    TRACE = 1;
-    // Debugging information.
-    DEBUG = 2;
-    // Normal events.
-    INFO = 3;
-    // Normal but significant events, such as start up, shut down, or
-    // configuration.
-    NOTICE = 4;
-    // Warning events might cause problems.
-    WARN = 5;
-    // Error events are likely to cause problems.
-    ERROR = 6;
-    // Critical events cause severe problems or brief outages and may
-    // indicate that a person must take action.
-    CRITICAL = 7;
+  message Severity {
+    enum Enum {
+      UNSPECIFIED = 0;
+      // Trace level information, also the default log level unless
+      // another severity is specified.
+      TRACE = 1;
+      // Debugging information.
+      DEBUG = 2;
+      // Normal events.
+      INFO = 3;
+      // Normal but significant events, such as start up, shut down, or
+      // configuration.
+      NOTICE = 4;
+      // Warning events might cause problems.
+      WARN = 5;
+      // Error events are likely to cause problems.
+      ERROR = 6;
+      // Critical events cause severe problems or brief outages and may
+      // indicate that a person must take action.
+      CRITICAL = 7;
+    }
   }
 
   // (Required) The severity of the log statement.
-  Severity severity = 1;
+  Severity.Enum severity = 1;
 
   // (Required) The time at which this log statement occurred.
   google.protobuf.Timestamp timestamp = 2;

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/common/runner-api/src/main/proto/beam_job_api.proto
----------------------------------------------------------------------
diff --git a/sdks/common/runner-api/src/main/proto/beam_job_api.proto b/sdks/common/runner-api/src/main/proto/beam_job_api.proto
index 9d826ff..d76e907 100644
--- a/sdks/common/runner-api/src/main/proto/beam_job_api.proto
+++ b/sdks/common/runner-api/src/main/proto/beam_job_api.proto
@@ -91,7 +91,7 @@ message RunJobResponse {
 }
 
 
-// Cancel is a synchronus request that returns a jobState back
+// Cancel is a synchronus request that returns a job state back
 // Throws error GRPC_STATUS_UNAVAILABLE if server is down
 // Throws error NOT_FOUND if the jobId is not found
 message CancelJobRequest {
@@ -101,11 +101,11 @@ message CancelJobRequest {
 
 // Valid responses include any terminal state or CANCELLING
 message CancelJobResponse {
-  JobState.JobStateType state = 1; // (required)
+  JobState.Enum state = 1; // (required)
 }
 
 
-// GetState is a synchronus request that returns a jobState back
+// GetState is a synchronus request that returns a job state back
 // Throws error GRPC_STATUS_UNAVAILABLE if server is down
 // Throws error NOT_FOUND if the jobId is not found
 message GetJobStateRequest {
@@ -114,7 +114,7 @@ message GetJobStateRequest {
 }
 
 message GetJobStateResponse {
-  JobState.JobStateType state = 1; // (required)
+  JobState.Enum state = 1; // (required)
 }
 
 
@@ -150,20 +150,19 @@ message JobMessagesResponse {
   }
 }
 
+// Enumeration of all JobStates
 message JobState {
-  // Enumeration of all JobStates
-  enum JobStateType {
-    JOB_STATE_TYPE_UNSPECIFIED = 0;
-    UNKNOWN = 1;
-    STOPPED = 2;
-    RUNNING = 3;
-    DONE = 4;
-    FAILED = 5;
-    CANCELLED = 6;
-    UPDATED = 7;
-    DRAINING = 8;
-    DRAINED = 9;
-    STARTING = 10;
-    CANCELLING = 11;
+  enum Enum {
+    UNSPECIFIED = 0;
+    STOPPED = 1;
+    RUNNING = 2;
+    DONE = 3;
+    FAILED = 4;
+    CANCELLED = 5;
+    UPDATED = 6;
+    DRAINING = 7;
+    DRAINED = 8;
+    STARTING = 9;
+    CANCELLING = 10;
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/common/runner-api/src/main/proto/beam_runner_api.proto
----------------------------------------------------------------------
diff --git a/sdks/common/runner-api/src/main/proto/beam_runner_api.proto b/sdks/common/runner-api/src/main/proto/beam_runner_api.proto
index 3b68993..9ba5577 100644
--- a/sdks/common/runner-api/src/main/proto/beam_runner_api.proto
+++ b/sdks/common/runner-api/src/main/proto/beam_runner_api.proto
@@ -191,7 +191,7 @@ message PCollection {
   string coder_id = 2;
 
   // (Required) Whether this PCollection is bounded or unbounded
-  IsBounded is_bounded = 3;
+  IsBounded.Enum is_bounded = 3;
 
   // (Required) The id of the windowing strategy for this PCollection.
   string windowing_strategy_id = 4;
@@ -242,13 +242,15 @@ message ParDoPayload {
 // TODO: the evolution of the Fn API will influence what needs explicit
 // representation here
 message Parameter {
-  Type type = 1;
-
-  enum Type {
-    TYPE_UNSPECIFIED = 0;
-    WINDOW = 1;
-    PIPELINE_OPTIONS = 2;
-    RESTRICTION_TRACKER = 3;
+  Type.Enum type = 1;
+
+  message Type {
+    enum Enum {
+      UNSPECIFIED = 0;
+      WINDOW = 1;
+      PIPELINE_OPTIONS = 2;
+      RESTRICTION_TRACKER = 3;
+    }
   }
 }
 
@@ -285,13 +287,15 @@ message SetStateSpec {
 }
 
 message TimerSpec {
-  TimeDomain time_domain = 1;
+  TimeDomain.Enum time_domain = 1;
 }
 
-enum IsBounded {
-  IS_BOUNDED_UNSPECIFIED = 0;
-  UNBOUNDED = 1;
-  BOUNDED = 2;
+message IsBounded {
+  enum Enum {
+    UNSPECIFIED = 0;
+    UNBOUNDED = 1;
+    BOUNDED = 2;
+  }
 }
 
 // The payload for the primitive Read transform.
@@ -301,7 +305,7 @@ message ReadPayload {
   SdkFunctionSpec source = 1;
 
   // (Required) Whether the source is bounded or unbounded
-  IsBounded is_bounded = 2;
+  IsBounded.Enum is_bounded = 2;
 
   // TODO: full audit of fields required by runners as opposed to SDK harness
 }
@@ -412,7 +416,7 @@ message WindowingStrategy {
   // (Required) Whether or not the window fn is merging.
   //
   // This knowledge is required for many optimizations.
-  MergeStatus merge_status = 2;
+  MergeStatus.Enum merge_status = 2;
 
   // (Required) The coder for the windows of this PCollection.
   string window_coder_id = 3;
@@ -424,7 +428,7 @@ message WindowingStrategy {
   // replacement for prior panes or whether they are deltas to be combined
   // with other panes (the combine should correspond to whatever the upstream
   // grouping transform is).
-  AccumulationMode accumulation_mode = 5;
+  AccumulationMode.Enum accumulation_mode = 5;
 
   // (Required) The OutputTime specifies, for a grouping transform, how to
   // compute the aggregate timestamp. The window_fn will first possibly shift
@@ -434,17 +438,17 @@ message WindowingStrategy {
   // This is actually only for input to grouping transforms, but since they
   // may be introduced in runner-specific ways, it is carried along with the
   // windowing strategy.
-  OutputTime output_time = 6;
+  OutputTime.Enum output_time = 6;
 
   // (Required) Indicate when output should be omitted upon window expiration.
-  ClosingBehavior closing_behavior = 7;
+  ClosingBehavior.Enum closing_behavior = 7;
 
   // (Required) The duration, in milliseconds, beyond the end of a window at
   // which the window becomes droppable.
   int64 allowed_lateness = 8;
 
   // (Required) Indicate whether empty on-time panes should be omitted.
-  OnTimeBehavior OnTimeBehavior = 9;
+  OnTimeBehavior.Enum OnTimeBehavior = 9;
 
   // (Required) Whether or not the window fn assigns inputs to exactly one window
   //
@@ -455,97 +459,109 @@ message WindowingStrategy {
 // Whether or not a PCollection's WindowFn is non-merging, merging, or
 // merging-but-already-merged, in which case a subsequent GroupByKey is almost
 // always going to do something the user does not want
-enum MergeStatus {
-  MERGE_STATUS_UNSPECIFIED = 0;
-
-  // The WindowFn does not require merging.
-  // Examples: global window, FixedWindows, SlidingWindows
-  NON_MERGING = 1;
-
-  // The WindowFn is merging and the PCollection has not had merging
-  // performed.
-  // Example: Sessions prior to a GroupByKey
-  NEEDS_MERGE = 2;
-
-  // The WindowFn is merging and the PCollection has had merging occur
-  // already.
-  // Example: Sessions after a GroupByKey
-  ALREADY_MERGED = 3;
+message MergeStatus {
+  enum Enum {
+    UNSPECIFIED = 0;
+
+    // The WindowFn does not require merging.
+    // Examples: global window, FixedWindows, SlidingWindows
+    NON_MERGING = 1;
+
+    // The WindowFn is merging and the PCollection has not had merging
+    // performed.
+    // Example: Sessions prior to a GroupByKey
+    NEEDS_MERGE = 2;
+
+    // The WindowFn is merging and the PCollection has had merging occur
+    // already.
+    // Example: Sessions after a GroupByKey
+    ALREADY_MERGED = 3;
+  }
 }
 
 // Whether or not subsequent outputs of aggregations should be entire
 // replacement values or just the aggregation of inputs received since
 // the prior output.
-enum AccumulationMode {
-  ACCUMULATION_MODE_UNSPECIFIED = 0;
+message AccumulationMode {
+  enum Enum {
+    UNSPECIFIED = 0;
 
-  // The aggregation is discarded when it is output
-  DISCARDING = 1;
+    // The aggregation is discarded when it is output
+    DISCARDING = 1;
 
-  // The aggregation is accumulated across outputs
-  ACCUMULATING = 2;
+    // The aggregation is accumulated across outputs
+    ACCUMULATING = 2;
+  }
 }
 
 // Controls whether or not an aggregating transform should output data
 // when a window expires.
-enum ClosingBehavior {
-  CLOSING_BEHVAIOR_UNSPECIFIED = 0;
+message ClosingBehavior {
+  enum Enum {
+    UNSPECIFIED = 0;
 
-  // Emit output when a window expires, whether or not there has been
-  // any new data since the last output.
-  EMIT_ALWAYS = 1;
+    // Emit output when a window expires, whether or not there has been
+    // any new data since the last output.
+    EMIT_ALWAYS = 1;
 
-  // Only emit output when new data has arrives since the last output
-  EMIT_IF_NONEMPTY = 2;
+    // Only emit output when new data has arrives since the last output
+    EMIT_IF_NONEMPTY = 2;
+  }
 }
 
 // Controls whether or not an aggregating transform should output data
 // when an on-time pane is empty.
-enum OnTimeBehavior {
-  ON_TIME_BEHAVIOR_UNSPECIFIED = 0;
+message OnTimeBehavior {
+  enum Enum {
+    UNSPECIFIED = 0;
 
-  // Always fire the on-time pane. Even if there is no new data since
-  // the previous firing, an element will be produced.
-  FIRE_ALWAYS = 1;
+    // Always fire the on-time pane. Even if there is no new data since
+    // the previous firing, an element will be produced.
+    FIRE_ALWAYS = 1;
 
-  // Only fire the on-time pane if there is new data since the previous firing.
-  FIRE_IF_NONEMPTY = 2;
+    // Only fire the on-time pane if there is new data since the previous firing.
+    FIRE_IF_NONEMPTY = 2;
+  }
 }
 
 // When a number of windowed, timestamped inputs are aggregated, the timestamp
 // for the resulting output.
-enum OutputTime {
-  OUTPUT_TIME_UNSPECIFIED = 0;
+message OutputTime {
+  enum Enum {
+    UNSPECIFIED = 0;
 
-  // The output has the timestamp of the end of the window.
-  END_OF_WINDOW = 1;
+    // The output has the timestamp of the end of the window.
+    END_OF_WINDOW = 1;
 
-  // The output has the latest timestamp of the input elements since
-  // the last output.
-  LATEST_IN_PANE = 2;
+    // The output has the latest timestamp of the input elements since
+    // the last output.
+    LATEST_IN_PANE = 2;
 
-  // The output has the earliest timestamp of the input elements since
-  // the last output.
-  EARLIEST_IN_PANE = 3;
+    // The output has the earliest timestamp of the input elements since
+    // the last output.
+    EARLIEST_IN_PANE = 3;
+  }
 }
 
 // The different time domains in the Beam model.
-enum TimeDomain {
-  TIME_DOMAIN_UNSPECIFIED = 0;
-
-  // Event time is time from the perspective of the data
-  EVENT_TIME = 1;
-
-  // Processing time is time from the perspective of the
-  // execution of your pipeline
-  PROCESSING_TIME = 2;
-
-  // Synchronized processing time is the minimum of the
-  // processing time of all pending elements.
-  //
-  // The "processing time" of an element refers to
-  // the local processing time at which it was emitted
-  SYNCHRONIZED_PROCESSING_TIME = 3;
+message TimeDomain {
+  enum Enum {
+    UNSPECIFIED = 0;
+
+    // Event time is time from the perspective of the data
+    EVENT_TIME = 1;
+
+    // Processing time is time from the perspective of the
+    // execution of your pipeline
+    PROCESSING_TIME = 2;
+
+    // Synchronized processing time is the minimum of the
+    // processing time of all pending elements.
+    //
+    // The "processing time" of an element refers to
+    // the local processing time at which it was emitted
+    SYNCHRONIZED_PROCESSING_TIME = 3;
+  }
 }
 
 // A small DSL for expressing when to emit new aggregations
@@ -799,7 +815,7 @@ message DisplayData {
     Identifier id = 1;
 
     // (Required)
-    Type type = 2;
+    Type.Enum type = 2;
 
     // (Required)
     google.protobuf.Any value = 3;
@@ -814,14 +830,16 @@ message DisplayData {
     string link_url = 6;
   }
 
-  enum Type {
-    TYPE_UNSPECIFIED = 0;
-    STRING = 1;
-    INTEGER = 2;
-    FLOAT = 3;
-    BOOLEAN = 4;
-    TIMESTAMP = 5;
-    DURATION = 6;
-    JAVA_CLASS = 7;
+  message Type {
+    enum Enum {
+      UNSPECIFIED = 0;
+      STRING = 1;
+      INTEGER = 2;
+      FLOAT = 3;
+      BOOLEAN = 4;
+      TIMESTAMP = 5;
+      DURATION = 6;
+      JAVA_CLASS = 7;
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
----------------------------------------------------------------------
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
index d56ee6d..c9f5d80 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
@@ -57,13 +57,13 @@ import org.apache.beam.sdk.options.PipelineOptions;
  */
 public class BeamFnLoggingClient implements AutoCloseable {
   private static final String ROOT_LOGGER_NAME = "";
-  private static final ImmutableMap<Level, BeamFnApi.LogEntry.Severity> LOG_LEVEL_MAP
=
-      ImmutableMap.<Level, BeamFnApi.LogEntry.Severity>builder()
-      .put(Level.SEVERE, BeamFnApi.LogEntry.Severity.ERROR)
-      .put(Level.WARNING, BeamFnApi.LogEntry.Severity.WARN)
-      .put(Level.INFO, BeamFnApi.LogEntry.Severity.INFO)
-      .put(Level.FINE, BeamFnApi.LogEntry.Severity.DEBUG)
-      .put(Level.FINEST, BeamFnApi.LogEntry.Severity.TRACE)
+  private static final ImmutableMap<Level, BeamFnApi.LogEntry.Severity.Enum> LOG_LEVEL_MAP
=
+      ImmutableMap.<Level, BeamFnApi.LogEntry.Severity.Enum>builder()
+      .put(Level.SEVERE, BeamFnApi.LogEntry.Severity.Enum.ERROR)
+      .put(Level.WARNING, BeamFnApi.LogEntry.Severity.Enum.WARN)
+      .put(Level.INFO, BeamFnApi.LogEntry.Severity.Enum.INFO)
+      .put(Level.FINE, BeamFnApi.LogEntry.Severity.Enum.DEBUG)
+      .put(Level.FINEST, BeamFnApi.LogEntry.Severity.Enum.TRACE)
       .build();
 
   private static final ImmutableMap<DataflowWorkerLoggingOptions.Level, Level> LEVEL_CONFIGURATION
=
@@ -190,7 +190,7 @@ public class BeamFnLoggingClient implements AutoCloseable {
 
     @Override
     public void publish(LogRecord record) {
-      BeamFnApi.LogEntry.Severity severity = LOG_LEVEL_MAP.get(record.getLevel());
+      BeamFnApi.LogEntry.Severity.Enum severity = LOG_LEVEL_MAP.get(record.getLevel());
       if (severity == null) {
         return;
       }

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClientTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClientTest.java
b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClientTest.java
index bb6a501..c2c26e7 100644
--- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClientTest.java
+++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClientTest.java
@@ -74,7 +74,7 @@ public class BeamFnLoggingClientTest {
 
   private static final BeamFnApi.LogEntry TEST_ENTRY =
       BeamFnApi.LogEntry.newBuilder()
-          .setSeverity(BeamFnApi.LogEntry.Severity.DEBUG)
+          .setSeverity(BeamFnApi.LogEntry.Severity.Enum.DEBUG)
           .setMessage("Message")
           .setThread("12345")
           .setTimestamp(Timestamp.newBuilder().setSeconds(1234567).setNanos(890000000).build())
@@ -82,7 +82,7 @@ public class BeamFnLoggingClientTest {
           .build();
   private static final BeamFnApi.LogEntry TEST_ENTRY_WITH_EXCEPTION =
       BeamFnApi.LogEntry.newBuilder()
-          .setSeverity(BeamFnApi.LogEntry.Severity.WARN)
+          .setSeverity(BeamFnApi.LogEntry.Severity.Enum.WARN)
           .setMessage("MessageWithException")
           .setTrace(getStackTraceAsString(TEST_RECORD_WITH_EXCEPTION.getThrown()))
           .setThread("12345")

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/io/iobase.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/iobase.py b/sdks/python/apache_beam/io/iobase.py
index 043666d..1f2a8bf 100644
--- a/sdks/python/apache_beam/io/iobase.py
+++ b/sdks/python/apache_beam/io/iobase.py
@@ -834,9 +834,9 @@ class Read(ptransform.PTransform):
     return (urns.READ_TRANSFORM,
             beam_runner_api_pb2.ReadPayload(
                 source=self.source.to_runner_api(context),
-                is_bounded=beam_runner_api_pb2.BOUNDED
+                is_bounded=beam_runner_api_pb2.IsBounded.BOUNDED
                 if self.source.is_bounded()
-                else beam_runner_api_pb2.UNBOUNDED))
+                else beam_runner_api_pb2.IsBounded.UNBOUNDED))
 
   @staticmethod
   def from_runner_api_parameter(parameter, context):

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index 53a6121..d2d3653 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -133,7 +133,7 @@ class PCollection(PValue):
         unique_name='%d%s.%s' % (
             len(self.producer.full_label), self.producer.full_label, self.tag),
         coder_id=pickler.dumps(self.element_type),
-        is_bounded=beam_runner_api_pb2.BOUNDED,
+        is_bounded=beam_runner_api_pb2.IsBounded.BOUNDED,
         windowing_strategy_id=context.windowing_strategies.get_id(
             self.windowing))
 

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/runners/experimental/python_rpc_direct/python_rpc_direct_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/experimental/python_rpc_direct/python_rpc_direct_runner.py
b/sdks/python/apache_beam/runners/experimental/python_rpc_direct/python_rpc_direct_runner.py
index 85e3f75..84bed42 100644
--- a/sdks/python/apache_beam/runners/experimental/python_rpc_direct/python_rpc_direct_runner.py
+++ b/sdks/python/apache_beam/runners/experimental/python_rpc_direct/python_rpc_direct_runner.py
@@ -96,7 +96,7 @@ class PythonRPCDirectPipelineResult(PipelineResult):
       if message.HasField('stateResponse'):
         logging.info(
             'Current state of job: %s',
-            beam_job_api_pb2.JobState.JobStateType.Name(
+            beam_job_api_pb2.JobState.Enum.Name(
                 message.stateResponse.state))
       else:
         logging.info('Message %s', message.messageResponse)

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/runners/experimental/python_rpc_direct/server.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/experimental/python_rpc_direct/server.py b/sdks/python/apache_beam/runners/experimental/python_rpc_direct/server.py
index 1d07e71..4986dc4 100644
--- a/sdks/python/apache_beam/runners/experimental/python_rpc_direct/server.py
+++ b/sdks/python/apache_beam/runners/experimental/python_rpc_direct/server.py
@@ -72,7 +72,7 @@ class JobService(beam_job_api_pb2_grpc.JobServiceServicer):
   @staticmethod
   def _map_state_to_jobState(state):
     if state == PipelineState.UNKNOWN:
-      return beam_job_api_pb2.JobState.UNKNOWN
+      return beam_job_api_pb2.JobState.UNSPECIFIED
     elif state == PipelineState.STOPPED:
       return beam_job_api_pb2.JobState.STOPPED
     elif state == PipelineState.RUNNING:

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/runners/portability/universal_local_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/portability/universal_local_runner.py b/sdks/python/apache_beam/runners/portability/universal_local_runner.py
index 844b3a8..bc62823 100644
--- a/sdks/python/apache_beam/runners/portability/universal_local_runner.py
+++ b/sdks/python/apache_beam/runners/portability/universal_local_runner.py
@@ -148,7 +148,7 @@ class UniversalLocalRunner(runner.PipelineRunner):
 
 class PipelineResult(runner.PipelineResult):
   def __init__(self, job_service, job_id):
-    super(PipelineResult, self).__init__(beam_job_api_pb2.JobState.UNKNOWN)
+    super(PipelineResult, self).__init__(beam_job_api_pb2.JobState.UNSPECIFIED)
     self._job_service = job_service
     self._job_id = job_id
     self._messages = []
@@ -167,11 +167,11 @@ class PipelineResult(runner.PipelineResult):
   def _runner_api_state_to_pipeline_state(runner_api_state):
     return getattr(
         runner.PipelineState,
-        beam_job_api_pb2.JobState.JobStateType.Name(runner_api_state))
+        beam_job_api_pb2.JobState.Enum.Name(runner_api_state))
 
   @staticmethod
   def _pipeline_state_to_runner_api_state(pipeline_state):
-    return beam_job_api_pb2.JobState.JobStateType.Value(pipeline_state)
+    return beam_job_api_pb2.JobState.Enum.Value(pipeline_state)
 
   def wait_until_finish(self):
     def read_messages():

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/runners/worker/log_handler.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/worker/log_handler.py b/sdks/python/apache_beam/runners/worker/log_handler.py
index 20bd49f..f878943 100644
--- a/sdks/python/apache_beam/runners/worker/log_handler.py
+++ b/sdks/python/apache_beam/runners/worker/log_handler.py
@@ -38,11 +38,11 @@ class FnApiLogRecordHandler(logging.Handler):
 
   # Mapping from logging levels to LogEntry levels.
   LOG_LEVEL_MAP = {
-      logging.FATAL: beam_fn_api_pb2.LogEntry.CRITICAL,
-      logging.ERROR: beam_fn_api_pb2.LogEntry.ERROR,
-      logging.WARNING: beam_fn_api_pb2.LogEntry.WARN,
-      logging.INFO: beam_fn_api_pb2.LogEntry.INFO,
-      logging.DEBUG: beam_fn_api_pb2.LogEntry.DEBUG
+      logging.FATAL: beam_fn_api_pb2.LogEntry.Severity.CRITICAL,
+      logging.ERROR: beam_fn_api_pb2.LogEntry.Severity.ERROR,
+      logging.WARNING: beam_fn_api_pb2.LogEntry.Severity.WARN,
+      logging.INFO: beam_fn_api_pb2.LogEntry.Severity.INFO,
+      logging.DEBUG: beam_fn_api_pb2.LogEntry.Severity.DEBUG
   }
 
   def __init__(self, log_service_descriptor):

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/runners/worker/log_handler_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/worker/log_handler_test.py b/sdks/python/apache_beam/runners/worker/log_handler_test.py
index 7edf667..e4323d2 100644
--- a/sdks/python/apache_beam/runners/worker/log_handler_test.py
+++ b/sdks/python/apache_beam/runners/worker/log_handler_test.py
@@ -73,7 +73,8 @@ class FnApiLogRecordHandlerTest(unittest.TestCase):
     num_received_log_entries = 0
     for outer in self.test_logging_service.log_records_received:
       for log_entry in outer.log_entries:
-        self.assertEqual(beam_fn_api_pb2.LogEntry.INFO, log_entry.severity)
+        self.assertEqual(beam_fn_api_pb2.LogEntry.Severity.INFO,
+                         log_entry.severity)
         self.assertEqual('%s: %s' % (msg, num_received_log_entries),
                          log_entry.message)
         self.assertEqual(u'log_handler_test._verify_fn_log_handler',

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/transforms/core.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
index 0a82de2..5d92fe9 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -1385,16 +1385,16 @@ class Windowing(object):
     return beam_runner_api_pb2.WindowingStrategy(
         window_fn=self.windowfn.to_runner_api(context),
         # TODO(robertwb): Prohibit implicit multi-level merging.
-        merge_status=(beam_runner_api_pb2.NEEDS_MERGE
+        merge_status=(beam_runner_api_pb2.MergeStatus.NEEDS_MERGE
                       if self.windowfn.is_merging()
-                      else beam_runner_api_pb2.NON_MERGING),
+                      else beam_runner_api_pb2.MergeStatus.NON_MERGING),
         window_coder_id=context.coders.get_id(
             self.windowfn.get_window_coder()),
         trigger=self.triggerfn.to_runner_api(context),
         accumulation_mode=self.accumulation_mode,
         output_time=self.timestamp_combiner,
         # TODO(robertwb): Support EMIT_IF_NONEMPTY
-        closing_behavior=beam_runner_api_pb2.EMIT_ALWAYS,
+        closing_behavior=beam_runner_api_pb2.ClosingBehavior.EMIT_ALWAYS,
         allowed_lateness=0)
 
   @staticmethod

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/transforms/trigger.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py
index 3583e62..bd99401 100644
--- a/sdks/python/apache_beam/transforms/trigger.py
+++ b/sdks/python/apache_beam/transforms/trigger.py
@@ -58,8 +58,8 @@ __all__ = [
 class AccumulationMode(object):
   """Controls what to do with data when a trigger fires multiple times.
   """
-  DISCARDING = beam_runner_api_pb2.DISCARDING
-  ACCUMULATING = beam_runner_api_pb2.ACCUMULATING
+  DISCARDING = beam_runner_api_pb2.AccumulationMode.DISCARDING
+  ACCUMULATING = beam_runner_api_pb2.AccumulationMode.ACCUMULATING
   # TODO(robertwb): Provide retractions of previous outputs.
   # RETRACTING = 3
 

http://git-wip-us.apache.org/repos/asf/beam/blob/6fddd4ed/sdks/python/apache_beam/transforms/window.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/window.py b/sdks/python/apache_beam/transforms/window.py
index a025019..8c8bf33 100644
--- a/sdks/python/apache_beam/transforms/window.py
+++ b/sdks/python/apache_beam/transforms/window.py
@@ -86,9 +86,9 @@ __all__ = [
 class TimestampCombiner(object):
   """Determines how output timestamps of grouping operations are assigned."""
 
-  OUTPUT_AT_EOW = beam_runner_api_pb2.END_OF_WINDOW
-  OUTPUT_AT_EARLIEST = beam_runner_api_pb2.EARLIEST_IN_PANE
-  OUTPUT_AT_LATEST = beam_runner_api_pb2.LATEST_IN_PANE
+  OUTPUT_AT_EOW = beam_runner_api_pb2.OutputTime.END_OF_WINDOW
+  OUTPUT_AT_EARLIEST = beam_runner_api_pb2.OutputTime.EARLIEST_IN_PANE
+  OUTPUT_AT_LATEST = beam_runner_api_pb2.OutputTime.LATEST_IN_PANE
   # TODO(robertwb): Add this to the runner API or remove it.
   OUTPUT_AT_EARLIEST_TRANSFORMED = 'OUTPUT_AT_EARLIEST_TRANSFORMED'
 


Mime
View raw message