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: Removes inputProvider() and outputReceiver()
Date Wed, 01 Feb 2017 03:07:47 GMT
Repository: beam
Updated Branches:
  refs/heads/master 77b136603 -> 38208eaa4


Removes inputProvider() and outputReceiver()

Removes InputProvider itself too.
Does not remove OutputReceiver because it's used in @SplitRestriction
method.
Cleans up tests that looked at InputProvider/OutputReceiver parameters -
instead now they look at DoFn.ProcessContext parameter, and I improved
the formatting of parameter types too.


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

Branch: refs/heads/master
Commit: 7d787bddc80fa832576407c313b26d5436dfa393
Parents: 34b4a6d
Author: Eugene Kirpichov <kirpichov@google.com>
Authored: Fri Jan 27 14:56:56 2017 -0800
Committer: Eugene Kirpichov <kirpichov@google.com>
Committed: Fri Jan 27 15:00:19 2017 -0800

----------------------------------------------------------------------
 .../apache/beam/runners/core/DoFnAdapters.java  | 20 -----
 .../beam/runners/core/SimpleDoFnRunner.java     | 32 --------
 .../beam/runners/core/SplittableParDo.java      | 12 ---
 .../org/apache/beam/sdk/transforms/DoFn.java    |  6 --
 .../apache/beam/sdk/transforms/DoFnTester.java  | 12 ---
 .../reflect/ByteBuddyDoFnInvokerFactory.java    | 12 ---
 .../sdk/transforms/reflect/DoFnInvoker.java     | 18 -----
 .../sdk/transforms/reflect/DoFnSignature.java   | 56 -------------
 .../sdk/transforms/reflect/DoFnSignatures.java  | 82 +++++---------------
 .../beam/sdk/util/common/ReflectHelpers.java    | 16 ++--
 .../transforms/reflect/DoFnInvokersTest.java    | 26 -------
 .../DoFnSignaturesProcessElementTest.java       | 40 ++++------
 .../DoFnSignaturesSplittableDoFnTest.java       |  3 +-
 .../transforms/reflect/DoFnSignaturesTest.java  |  6 +-
 14 files changed, 51 insertions(+), 290 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
index 23aba58..dcd7969 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
@@ -204,16 +204,6 @@ public class DoFnAdapters {
     }
 
     @Override
-    public DoFn.InputProvider<InputT> inputProvider() {
-      throw new UnsupportedOperationException("inputProvider() exists only for testing");
-    }
-
-    @Override
-    public DoFn.OutputReceiver<OutputT> outputReceiver() {
-      throw new UnsupportedOperationException("outputReceiver() exists only for testing");
-    }
-
-    @Override
     public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
       throw new UnsupportedOperationException("This is a non-splittable DoFn");
     }
@@ -316,16 +306,6 @@ public class DoFnAdapters {
     }
 
     @Override
-    public DoFn.InputProvider<InputT> inputProvider() {
-      throw new UnsupportedOperationException("inputProvider() exists only for testing");
-    }
-
-    @Override
-    public DoFn.OutputReceiver<OutputT> outputReceiver() {
-      throw new UnsupportedOperationException("outputReceiver() exists only for testing");
-    }
-
-    @Override
     public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
       throw new UnsupportedOperationException("This is a non-splittable DoFn");
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
index df5f3f6..d54daf6 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
@@ -34,9 +34,7 @@ import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.Combine.CombineFn;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.DoFn.Context;
-import org.apache.beam.sdk.transforms.DoFn.InputProvider;
 import org.apache.beam.sdk.transforms.DoFn.OnTimerContext;
-import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
 import org.apache.beam.sdk.transforms.DoFn.ProcessContext;
 import org.apache.beam.sdk.transforms.reflect.DoFnInvoker;
 import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
@@ -439,16 +437,6 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     }
 
     @Override
-    public InputProvider<InputT> inputProvider() {
-      throw new UnsupportedOperationException("InputProvider is for testing only.");
-    }
-
-    @Override
-    public OutputReceiver<OutputT> outputReceiver() {
-      throw new UnsupportedOperationException("OutputReceiver is for testing only.");
-    }
-
-    @Override
     public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
       throw new UnsupportedOperationException(
           "Cannot access RestrictionTracker outside of @ProcessElement method.");
@@ -626,16 +614,6 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     }
 
     @Override
-    public InputProvider<InputT> inputProvider() {
-      throw new UnsupportedOperationException("InputProvider parameters are not supported.");
-    }
-
-    @Override
-    public OutputReceiver<OutputT> outputReceiver() {
-      throw new UnsupportedOperationException("OutputReceiver parameters are not supported.");
-    }
-
-    @Override
     public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
       throw new UnsupportedOperationException("RestrictionTracker parameters are not supported.");
     }
@@ -744,16 +722,6 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     }
 
     @Override
-    public InputProvider<InputT> inputProvider() {
-      throw new UnsupportedOperationException("InputProvider parameters are not supported.");
-    }
-
-    @Override
-    public OutputReceiver<OutputT> outputReceiver() {
-      throw new UnsupportedOperationException("OutputReceiver parameters are not supported.");
-    }
-
-    @Override
     public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
       throw new UnsupportedOperationException("RestrictionTracker parameters are not supported.");
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
index f8d12ec..d1cbf8f 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
@@ -672,18 +672,6 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
       }
 
       @Override
-      public DoFn.InputProvider<InputT> inputProvider() {
-        // DoFnSignatures should have verified that this DoFn doesn't access extra context.
-        throw new IllegalStateException("Unexpected extra context access on a splittable
DoFn");
-      }
-
-      @Override
-      public DoFn.OutputReceiver<OutputT> outputReceiver() {
-        // DoFnSignatures should have verified that this DoFn doesn't access extra context.
-        throw new IllegalStateException("Unexpected extra context access on a splittable
DoFn");
-      }
-
-      @Override
       public TrackerT restrictionTracker() {
         return tracker;
       }

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
index 699403f..a161919 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
@@ -372,12 +372,6 @@ public abstract class DoFn<InputT, OutputT> implements Serializable,
HasDisplayD
   public interface OutputReceiver<T> {
     void output(T output);
   }
-
-  /** Provides a single value of the given type. */
-  public interface InputProvider<T> {
-    T get();
-  }
-
   /////////////////////////////////////////////////////////////////////////////
 
   /**

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
index b2c3fd5..0d1f96d 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
@@ -315,18 +315,6 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable
{
             }
 
             @Override
-            public DoFn.InputProvider<InputT> inputProvider() {
-              throw new UnsupportedOperationException(
-                  "Not expected to access InputProvider from DoFnTester");
-            }
-
-            @Override
-            public DoFn.OutputReceiver<OutputT> outputReceiver() {
-              throw new UnsupportedOperationException(
-                  "Not expected to access OutputReceiver from DoFnTester");
-            }
-
-            @Override
             public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker()
{
               throw new UnsupportedOperationException(
                   "Not expected to access RestrictionTracker from a regular DoFn in DoFnTester");

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java
index 01ddd86..46b21d6 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java
@@ -68,9 +68,7 @@ import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.OnTimerMethod;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.Cases;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.ContextParameter;
-import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.InputProviderParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.OnTimerContextParameter;
-import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.OutputReceiverParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.ProcessContextParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.RestrictionTrackerParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.StateParameter;
@@ -572,16 +570,6 @@ public class ByteBuddyDoFnInvokerFactory implements DoFnInvokerFactory
{
           }
 
           @Override
-          public StackManipulation dispatch(InputProviderParameter p) {
-            return simpleExtraContextParameter(INPUT_PROVIDER_PARAMETER_METHOD);
-          }
-
-          @Override
-          public StackManipulation dispatch(OutputReceiverParameter p) {
-            return simpleExtraContextParameter(OUTPUT_RECEIVER_PARAMETER_METHOD);
-          }
-
-          @Override
           public StackManipulation dispatch(RestrictionTrackerParameter p) {
             // DoFnInvoker.ArgumentProvider.restrictionTracker() returns a RestrictionTracker,
             // but the @ProcessElement method expects a concrete subtype of it.

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java
index 354578e..5f61349 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java
@@ -21,8 +21,6 @@ import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderRegistry;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.DoFn.FinishBundle;
-import org.apache.beam.sdk.transforms.DoFn.InputProvider;
-import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
 import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
 import org.apache.beam.sdk.transforms.DoFn.StartBundle;
 import org.apache.beam.sdk.transforms.DoFn.StateId;
@@ -113,12 +111,6 @@ public interface DoFnInvoker<InputT, OutputT> {
     /** Provide a {@link DoFn.OnTimerContext} to use with the given {@link DoFn}. */
     DoFn<InputT, OutputT>.OnTimerContext onTimerContext(DoFn<InputT, OutputT>
doFn);
 
-    /** A placeholder for testing purposes. */
-    InputProvider<InputT> inputProvider();
-
-    /** A placeholder for testing purposes. */
-    OutputReceiver<OutputT> outputReceiver();
-
     /**
      * If this is a splittable {@link DoFn}, returns the {@link RestrictionTracker} associated
with
      * the current {@link ProcessElement} call.
@@ -155,16 +147,6 @@ public interface DoFnInvoker<InputT, OutputT> {
     }
 
     @Override
-    public InputProvider<InputT> inputProvider() {
-      return null;
-    }
-
-    @Override
-    public OutputReceiver<OutputT> outputReceiver() {
-      return null;
-    }
-
-    @Override
     public State state(String stateId) {
       return null;
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
index f470782..007d8be 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
@@ -28,8 +28,6 @@ import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.DoFn.InputProvider;
-import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
 import org.apache.beam.sdk.transforms.DoFn.ProcessContinuation;
 import org.apache.beam.sdk.transforms.DoFn.StateId;
 import org.apache.beam.sdk.transforms.DoFn.TimerId;
@@ -190,10 +188,6 @@ public abstract class DoFnSignature {
         return cases.dispatch((WindowParameter) this);
       } else if (this instanceof RestrictionTrackerParameter) {
         return cases.dispatch((RestrictionTrackerParameter) this);
-      } else if (this instanceof InputProviderParameter) {
-        return cases.dispatch((InputProviderParameter) this);
-      } else if (this instanceof OutputReceiverParameter) {
-        return cases.dispatch((OutputReceiverParameter) this);
       } else if (this instanceof StateParameter) {
         return cases.dispatch((StateParameter) this);
       } else if (this instanceof TimerParameter) {
@@ -213,8 +207,6 @@ public abstract class DoFnSignature {
       ResultT dispatch(ProcessContextParameter p);
       ResultT dispatch(OnTimerContextParameter p);
       ResultT dispatch(WindowParameter p);
-      ResultT dispatch(InputProviderParameter p);
-      ResultT dispatch(OutputReceiverParameter p);
       ResultT dispatch(RestrictionTrackerParameter p);
       ResultT dispatch(StateParameter p);
       ResultT dispatch(TimerParameter p);
@@ -247,16 +239,6 @@ public abstract class DoFnSignature {
         }
 
         @Override
-        public ResultT dispatch(InputProviderParameter p) {
-          return dispatchDefault(p);
-        }
-
-        @Override
-        public ResultT dispatch(OutputReceiverParameter p) {
-          return dispatchDefault(p);
-        }
-
-        @Override
         public ResultT dispatch(RestrictionTrackerParameter p) {
           return dispatchDefault(p);
         }
@@ -280,10 +262,6 @@ public abstract class DoFnSignature {
           new AutoValue_DoFnSignature_Parameter_ProcessContextParameter();
     private static final OnTimerContextParameter ON_TIMER_CONTEXT_PARAMETER =
         new AutoValue_DoFnSignature_Parameter_OnTimerContextParameter();
-    private static final InputProviderParameter INPUT_PROVIDER_PARAMETER =
-        new AutoValue_DoFnSignature_Parameter_InputProviderParameter();
-    private static final OutputReceiverParameter OUTPUT_RECEIVER_PARAMETER =
-        new AutoValue_DoFnSignature_Parameter_OutputReceiverParameter();
 
     /** Returns a {@link ContextParameter}. */
     public static ContextParameter context() {
@@ -306,20 +284,6 @@ public abstract class DoFnSignature {
     }
 
     /**
-     * Returns an {@link InputProviderParameter}.
-     */
-    public static InputProviderParameter inputProvider() {
-      return INPUT_PROVIDER_PARAMETER;
-    }
-
-    /**
-     * Returns an {@link OutputReceiverParameter}.
-     */
-    public static OutputReceiverParameter outputReceiver() {
-      return OUTPUT_RECEIVER_PARAMETER;
-    }
-
-    /**
      * Returns a {@link RestrictionTrackerParameter}.
      */
     public static RestrictionTrackerParameter restrictionTracker(TypeDescriptor<?>
trackerT) {
@@ -378,26 +342,6 @@ public abstract class DoFnSignature {
     }
 
     /**
-     * Descriptor for a {@link Parameter} of type {@link InputProvider}.
-     *
-     * <p>All such descriptors are equal.
-     */
-    @AutoValue
-    public abstract static class InputProviderParameter extends Parameter {
-      InputProviderParameter() {}
-    }
-
-    /**
-     * Descriptor for a {@link Parameter} of type {@link OutputReceiver}.
-     *
-     * <p>All such descriptors are equal.
-     */
-    @AutoValue
-    public abstract static class OutputReceiverParameter extends Parameter {
-      OutputReceiverParameter() {}
-    }
-
-    /**
      * Descriptor for a {@link Parameter} of a subclass of {@link RestrictionTracker}.
      *
      * <p>All such descriptors are equal.

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
index b6b764e..61b9157 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
@@ -80,9 +80,7 @@ public class DoFnSignatures {
           Parameter.ProcessContextParameter.class,
           Parameter.WindowParameter.class,
           Parameter.TimerParameter.class,
-          Parameter.StateParameter.class,
-          Parameter.InputProviderParameter.class,
-          Parameter.OutputReceiverParameter.class);
+          Parameter.StateParameter.class);
 
   private static final Collection<Class<? extends Parameter>>
       ALLOWED_SPLITTABLE_PROCESS_ELEMENT_PARAMETERS =
@@ -617,25 +615,6 @@ public class DoFnSignatures {
         .where(new TypeParameter<OutputT>() {}, outputT);
   }
 
-  /**
-   * Generates a {@link TypeDescriptor} for {@code DoFn.InputProvider<InputT>} given
{@code InputT}.
-   */
-  private static <InputT> TypeDescriptor<DoFn.InputProvider<InputT>> inputProviderTypeOf(
-      TypeDescriptor<InputT> inputT) {
-    return new TypeDescriptor<DoFn.InputProvider<InputT>>() {}.where(
-        new TypeParameter<InputT>() {}, inputT);
-  }
-
-  /**
-   * Generates a {@link TypeDescriptor} for {@code DoFn.OutputReceiver<OutputT>} given
{@code
-   * OutputT}.
-   */
-  private static <OutputT> TypeDescriptor<DoFn.OutputReceiver<OutputT>>
outputReceiverTypeOf(
-      TypeDescriptor<OutputT> inputT) {
-    return new TypeDescriptor<DoFn.OutputReceiver<OutputT>>() {}.where(
-        new TypeParameter<OutputT>() {}, inputT);
-  }
-
   @VisibleForTesting
   static DoFnSignature.OnTimerMethod analyzeOnTimerMethod(
       ErrorReporter errors,
@@ -767,8 +746,6 @@ public class DoFnSignatures {
     TypeDescriptor<?> expectedProcessContextT = doFnProcessContextTypeOf(inputT, outputT);
     TypeDescriptor<?> expectedContextT = doFnContextTypeOf(inputT, outputT);
     TypeDescriptor<?> expectedOnTimerContextT = doFnOnTimerContextTypeOf(inputT, outputT);
-    TypeDescriptor<?> expectedInputProviderT = inputProviderTypeOf(inputT);
-    TypeDescriptor<?> expectedOutputReceiverT = outputReceiverTypeOf(outputT);
 
     TypeDescriptor<?> paramT = param.getType();
     Class<?> rawType = paramT.getRawType();
@@ -776,51 +753,27 @@ public class DoFnSignatures {
     ErrorReporter paramErrors = methodErrors.forParameter(param);
 
     if (rawType.equals(DoFn.ProcessContext.class)) {
-      methodErrors.checkArgument(paramT.equals(expectedProcessContextT),
-        "Must take %s as the ProcessContext argument",
+      paramErrors.checkArgument(paramT.equals(expectedProcessContextT),
+        "ProcessContext argument must have type %s",
         formatType(expectedProcessContextT));
       return Parameter.processContext();
     } else if (rawType.equals(DoFn.Context.class)) {
-      methodErrors.checkArgument(paramT.equals(expectedContextT),
-          "Must take %s as the Context argument",
+      paramErrors.checkArgument(paramT.equals(expectedContextT),
+          "Context argument must have type %s",
           formatType(expectedContextT));
       return Parameter.context();
     } else if (rawType.equals(DoFn.OnTimerContext.class)) {
-        methodErrors.checkArgument(paramT.equals(expectedOnTimerContextT),
-            "Must take %s as the OnTimerContext argument",
-            formatType(expectedOnTimerContextT));
-        return Parameter.onTimerContext();
+      paramErrors.checkArgument(
+          paramT.equals(expectedOnTimerContextT),
+          "OnTimerContext argument must have type %s",
+          formatType(expectedOnTimerContextT));
+      return Parameter.onTimerContext();
     } else if (BoundedWindow.class.isAssignableFrom(rawType)) {
       methodErrors.checkArgument(
           !methodContext.hasWindowParameter(),
           "Multiple %s parameters",
           BoundedWindow.class.getSimpleName());
       return Parameter.boundedWindow((TypeDescriptor<? extends BoundedWindow>) paramT);
-    } else if (rawType.equals(DoFn.InputProvider.class)) {
-      methodErrors.checkArgument(
-          !methodContext.getExtraParameters().contains(Parameter.inputProvider()),
-          "Multiple %s parameters",
-          DoFn.InputProvider.class.getSimpleName());
-      paramErrors.checkArgument(
-          paramT.equals(expectedInputProviderT),
-          "%s is for %s when it should be %s",
-          DoFn.InputProvider.class.getSimpleName(),
-          formatType(paramT),
-          formatType(expectedInputProviderT));
-      return Parameter.inputProvider();
-    } else if (rawType.equals(DoFn.OutputReceiver.class)) {
-      methodErrors.checkArgument(
-          !methodContext.getExtraParameters().contains(Parameter.outputReceiver()),
-          "Multiple %s parameters",
-          DoFn.OutputReceiver.class.getSimpleName());
-      paramErrors.checkArgument(
-          paramT.equals(expectedOutputReceiverT),
-          "%s is for %s when it should be %s",
-          DoFn.OutputReceiver.class.getSimpleName(),
-          formatType(paramT),
-          formatType(expectedOutputReceiverT));
-      return Parameter.outputReceiver();
-
     } else if (RestrictionTracker.class.isAssignableFrom(rawType)) {
       methodErrors.checkArgument(
           !methodContext.hasRestrictionTrackerParameter(),
@@ -890,7 +843,7 @@ public class DoFnSignatures {
           "reference to %s %s with different type %s",
           StateId.class.getSimpleName(),
           id,
-          stateDecl.stateType());
+          formatType(stateDecl.stateType()));
 
       paramErrors.checkArgument(
           stateDecl.field().getDeclaringClass().equals(param.getMethod().getDeclaringClass()),
@@ -1001,9 +954,14 @@ public class DoFnSignatures {
         m, fnT.resolveType(m.getGenericReturnType()));
   }
 
-  /** Generates a {@link TypeDescriptor} for {@code List<T>} given {@code T}. */
-  private static <T> TypeDescriptor<List<T>> listTypeOf(TypeDescriptor<T>
elementT) {
-    return new TypeDescriptor<List<T>>() {}.where(new TypeParameter<T>()
{}, elementT);
+  /**
+   * Generates a {@link TypeDescriptor} for {@code DoFn.OutputReceiver<OutputT>} given
{@code
+   * OutputT}.
+   */
+  private static <OutputT> TypeDescriptor<DoFn.OutputReceiver<OutputT>>
outputReceiverTypeOf(
+      TypeDescriptor<OutputT> inputT) {
+    return new TypeDescriptor<DoFn.OutputReceiver<OutputT>>() {}.where(
+        new TypeParameter<OutputT>() {}, inputT);
   }
 
   @VisibleForTesting
@@ -1317,7 +1275,7 @@ public class DoFnSignatures {
           this,
           String.format(
               "parameter of type %s at index %s",
-              param.getType(), param.getIndex()));
+              formatType(param.getType()), param.getIndex()));
     }
 
     void throwIllegalArgument(String message, Object... args) {

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ReflectHelpers.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ReflectHelpers.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ReflectHelpers.java
index 4ec39c1..7752b2a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ReflectHelpers.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ReflectHelpers.java
@@ -155,12 +155,18 @@ public class ReflectHelpers {
     }
 
     private void formatParameterizedType(StringBuilder builder, ParameterizedType t) {
+      if (t.getOwnerType() != null) {
+        format(builder, t.getOwnerType());
+        builder.append('.');
+      }
       format(builder, t.getRawType());
-      builder.append('<');
-      COMMA_SEPARATOR.appendTo(builder,
-          FluentIterable.from(asList(t.getActualTypeArguments()))
-          .transform(TYPE_SIMPLE_DESCRIPTION));
-      builder.append('>');
+      if (t.getActualTypeArguments().length > 0) {
+        builder.append('<');
+        COMMA_SEPARATOR.appendTo(builder,
+            FluentIterable.from(asList(t.getActualTypeArguments()))
+                .transform(TYPE_SIMPLE_DESCRIPTION));
+        builder.append('>');
+      }
     }
 
     private void formatGenericArrayType(StringBuilder builder, GenericArrayType t) {

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
index 4c6bee1..9bc2d12 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
@@ -73,16 +73,12 @@ public class DoFnInvokersTest {
 
   @Mock private DoFn<String, String>.ProcessContext mockProcessContext;
   @Mock private IntervalWindow mockWindow;
-  @Mock private DoFn.InputProvider<String> mockInputProvider;
-  @Mock private DoFn.OutputReceiver<String> mockOutputReceiver;
   @Mock private DoFnInvoker.ArgumentProvider<String, String> mockArgumentProvider;
 
   @Before
   public void setUp() {
     MockitoAnnotations.initMocks(this);
     when(mockArgumentProvider.window()).thenReturn(mockWindow);
-    when(mockArgumentProvider.inputProvider()).thenReturn(mockInputProvider);
-    when(mockArgumentProvider.outputReceiver()).thenReturn(mockOutputReceiver);
     when(mockArgumentProvider.processContext(Matchers.<DoFn>any())).thenReturn(mockProcessContext);
   }
 
@@ -231,28 +227,6 @@ public class DoFnInvokersTest {
   }
 
   @Test
-  public void testDoFnWithOutputReceiver() throws Exception {
-    class MockFn extends DoFn<String, String> {
-      @DoFn.ProcessElement
-      public void processElement(ProcessContext c, OutputReceiver<String> o) throws
Exception {}
-    }
-    MockFn fn = mock(MockFn.class);
-    assertEquals(ProcessContinuation.stop(), invokeProcessElement(fn));
-    verify(fn).processElement(mockProcessContext, mockOutputReceiver);
-  }
-
-  @Test
-  public void testDoFnWithInputProvider() throws Exception {
-    class MockFn extends DoFn<String, String> {
-      @DoFn.ProcessElement
-      public void processElement(ProcessContext c, InputProvider<String> o) throws
Exception {}
-    }
-    MockFn fn = mock(MockFn.class);
-    assertEquals(ProcessContinuation.stop(), invokeProcessElement(fn));
-    verify(fn).processElement(mockProcessContext, mockInputProvider);
-  }
-
-  @Test
   public void testDoFnWithReturn() throws Exception {
     class MockFn extends DoFn<String, String> {
       @DoFn.ProcessElement

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesProcessElementTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesProcessElementTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesProcessElementTest.java
index 5255af8..44ae5c4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesProcessElementTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesProcessElementTest.java
@@ -65,41 +65,35 @@ public class DoFnSignaturesProcessElementTest {
     analyzeProcessElementMethod(
         new AnonymousMethod() {
           private void method(
-              DoFn<Integer, String>.ProcessContext c,
-              DoFn.InputProvider<Integer> input,
-              DoFn.OutputReceiver<String> output) {}
+              DoFn<Integer, String>.ProcessContext c) {}
         });
   }
 
   @Test
   public void testBadGenericsTwoArgs() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("OutputReceiver<Integer>");
-    thrown.expectMessage("should be");
-    thrown.expectMessage("OutputReceiver<String>");
+    thrown.expectMessage("DoFn<Integer, Integer>.ProcessContext");
+    thrown.expectMessage("must have type");
+    thrown.expectMessage("DoFn<Integer, String>.ProcessContext");
 
     analyzeProcessElementMethod(
         new AnonymousMethod() {
           private void method(
-              DoFn<Integer, String>.ProcessContext c,
-              DoFn.InputProvider<Integer> input,
-              DoFn.OutputReceiver<Integer> output) {}
+              DoFn<Integer, Integer>.ProcessContext c) {}
         });
   }
 
   @Test
   public void testBadGenericWildCards() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("OutputReceiver<? super Integer>");
-    thrown.expectMessage("should be");
-    thrown.expectMessage("OutputReceiver<String>");
+    thrown.expectMessage("DoFn<Integer, ? super Integer>.ProcessContext");
+    thrown.expectMessage("must have type");
+    thrown.expectMessage("DoFn<Integer, String>.ProcessContext");
 
     analyzeProcessElementMethod(
         new AnonymousMethod() {
           private void method(
-              DoFn<Integer, String>.ProcessContext c,
-              DoFn.InputProvider<Integer> input,
-              DoFn.OutputReceiver<? super Integer> output) {}
+              DoFn<Integer, ? super Integer>.ProcessContext c) {}
         });
   }
 
@@ -107,17 +101,15 @@ public class DoFnSignaturesProcessElementTest {
     @ProcessElement
     @SuppressWarnings("unused")
     public void badTypeVariables(
-        DoFn<InputT, OutputT>.ProcessContext c,
-        DoFn.InputProvider<InputT> input,
-        DoFn.OutputReceiver<InputT> output) {}
+        DoFn<InputT, InputT>.ProcessContext c) {}
   }
 
   @Test
   public void testBadTypeVariables() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("OutputReceiver<InputT>");
-    thrown.expectMessage("should be");
-    thrown.expectMessage("OutputReceiver<OutputT>");
+    thrown.expectMessage("DoFn<InputT, InputT>.ProcessContext");
+    thrown.expectMessage("must have type");
+    thrown.expectMessage("DoFn<InputT, OutputT>.ProcessContext");
 
     DoFnSignatures.getSignature(BadTypeVariables.class);
   }
@@ -164,9 +156,7 @@ public class DoFnSignaturesProcessElementTest {
     @ProcessElement
     @SuppressWarnings("unused")
     public void goodTypeVariables(
-        DoFn<InputT, OutputT>.ProcessContext c,
-        DoFn.InputProvider<InputT> input,
-        DoFn.OutputReceiver<OutputT> output) {}
+        DoFn<InputT, OutputT>.ProcessContext c) {}
   }
 
   @Test
@@ -177,7 +167,7 @@ public class DoFnSignaturesProcessElementTest {
   private static class IdentityFn<T> extends DoFn<T, T> {
     @ProcessElement
     @SuppressWarnings("unused")
-    public void processElement(ProcessContext c, InputProvider<T> input, OutputReceiver<T>
output) {
+    public void processElement(ProcessContext c) {
       c.output(c.element());
     }
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesSplittableDoFnTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesSplittableDoFnTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesSplittableDoFnTest.java
index 7b594c9..c10d199 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesSplittableDoFnTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesSplittableDoFnTest.java
@@ -403,7 +403,8 @@ public class DoFnSignaturesSplittableDoFnTest {
   @Test
   public void testSplitRestrictionReturnsWrongType() throws Exception {
     thrown.expectMessage(
-        "Third argument must be OutputReceiver<SomeRestriction>, but is OutputReceiver<String>");
+        "Third argument must be DoFn.OutputReceiver<SomeRestriction>, "
+            + "but is DoFn.OutputReceiver<String>");
     DoFnSignatures.analyzeSplitRestrictionMethod(
         errors(),
         TypeDescriptor.of(FakeDoFn.class),

http://git-wip-us.apache.org/repos/asf/beam/blob/7d787bdd/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
index 69d4058..e1fa2d1 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
@@ -76,7 +76,7 @@ public class DoFnSignaturesTest {
   @Test
   public void testBadExtraContext() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("Must take a single argument of type Context");
+    thrown.expectMessage("Must take a single argument of type DoFn<Integer, String>.Context");
 
     DoFnSignatures.analyzeBundleMethod(
         errors(),
@@ -656,10 +656,10 @@ public class DoFnSignaturesTest {
   @Test
   public void testStateParameterWrongGenericType() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("ValueState<java.lang.String>");
+    thrown.expectMessage("ValueState<String>");
     thrown.expectMessage("reference to");
     thrown.expectMessage("different type");
-    thrown.expectMessage("ValueState<java.lang.Integer>");
+    thrown.expectMessage("ValueState<Integer>");
     thrown.expectMessage("my-id");
     thrown.expectMessage("myProcessElement");
     thrown.expectMessage("index 1");


Mime
View raw message