title: Spark Java API Task not serializable when not using Lambda
body: <p>I am seeing a behavior in Spark ( 2.2.0 ) I do not understand, but guessing it's related to Lambda and Anonymous classes, when trying to extract out a lambda function:</p>
  
  <p>This works:</p>
  
  <pre><code>public class EventsFilter
  {
      public Dataset&lt; String &gt; filter( Dataset&lt; String &gt; events )
      {
          return events.filter( ( FilterFunction&lt; String &gt; ) x -&gt; x.length() &gt; 3 );
      }
  }
  </code></pre>
  
  <p>Yet this does not:</p>
  
  <pre><code>public class EventsFilter
  {
      public Dataset&lt; String &gt; filter( Dataset&lt; String &gt; events )
      {
          FilterFunction&lt; String &gt; filter = new FilterFunction&lt; String &gt;(){
              @Override public boolean call( String value ) throws Exception
              {
                  return value.length() &gt; 3;
              }
          };
          return events.filter( filter );
      }
  }
  
  org.apache.spark.SparkException: Task not serializable
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298) ...
  ...
  Caused by: java.io.NotSerializableException: ...EventsFilter
     ..Serialization stack:
  - object not serializable (class: ...EventsFilter, 
  value:...EventsFilter@e521067)
      - field (class: .EventsFilter$1, name: this$0, type: class ..EventsFilter)
  .   - object (class ...EventsFilter$1, ..EventsFilter$1@5c70d7f0)
  .   - element of array (index: 1)
      - array (class [Ljava.lang.Object;, size 4)
      - field (class: 
  org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8, name: references$1, type: class [Ljava.lang.Object;)
      - object (class org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8, &lt;function2&gt;)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
  at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295)
  </code></pre>
  
  <p>I am testing against:</p>
  
  <pre><code>@Test
  public void test()
  {
      EventsFilter filter = new EventsFilter();
      Dataset&lt;String&gt; input = SparkSession.builder().appName( "test" ).master( "local" ).getOrCreate()
              .createDataset( Arrays.asList( "123" , "123"  , "3211" ) ,
                  Encoders.kryo( String.class ) );
  
      Dataset&lt;String&gt; res = filter.filter( input );
      assertThat( res.count() , is( 1l ) );
  }
  </code></pre>
  
  <p>Even weirder, when put in a static main, both seem to work...</p>
  
  <p>How is defining the function explicitly inside a method causing that sneaky 'this' reference serialization? </p>