@@ -0,0 +1,2876 @@
+/*%0A * Licensed to the Apache Software Foundation (ASF) under one%0A * or more contributor license agreements.  See the NOTICE file%0A * distributed with this work for additional information%0A * regarding copyright ownership.  The ASF licenses this file%0A * to you under the Apache License, Version 2.0 (the%0A * %22License%22); you may not use this file except in compliance%0A * with the License.  You may obtain a copy of the License at%0A *%0A *     http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0A%0Apackage org.apache.beam.runners.spark.aggregators.metrics.sink;%0A%0Aimport static org.hamcrest.Matchers.is;%0Aimport static org.hamcrest.Matchers.nullValue;%0Aimport static org.junit.Assert.assertThat;%0A%0Aimport com.google.common.collect.ImmutableSet;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0Aimport java.util.Set;%0Aimport org.apache.beam.runners.spark.PipelineRule;%0Aimport org.apache.beam.runners.spark.examples.WordCount;%0Aimport org.apache.beam.sdk.Pipeline;%0Aimport org.apache.beam.sdk.coders.StringUtf8Coder;%0Aimport org.apache.beam.sdk.testing.PAssert;%0Aimport org.apache.beam.sdk.transforms.Create;%0Aimport org.apache.beam.sdk.transforms.MapElements;%0Aimport org.apache.beam.sdk.values.PCollection;%0Aimport org.junit.Rule;%0Aimport org.junit.Test;%0Aimport org.junit.rules.ExternalResource;%0A%0A%0A/**%0A * A test that verifies Beam metrics are reported to Spark's metrics sink.%0A */%0Apublic class SparkMetricsSinkTest %7B%0A%0A  @Rule%0A  public ExternalResource inMemoryMetricsSink = new InMemoryMetricsSinkRule();%0A%0A  @Rule%0A  public final PipelineRule pipelineRule = PipelineRule.batch();%0A%0A  private Pipeline createSparkPipeline() %7B%0A    pipelineRule.getOptions().setEnableSparkMetricSinks(true);%0A    return pipelineRule.createPipeline();%0A  %7D%0A%0A  private void runPipeline() %7B%0A    final List%3CString%3E words =%0A        Arrays.asList(%22hi there%22, %22hi%22, %22hi sue bob%22, %22hi sue%22, %22%22, %22bob hi%22);%0A%0A    final Set%3CString%3E expectedCounts =%0A        ImmutableSet.of(%22hi: 5%22, %22there: 1%22, %22sue: 2%22, %22bob: 2%22);%0A%0A    final Pipeline pipeline = createSparkPipeline();%0A%0A    final PCollection%3CString%3E output =%0A        pipeline%0A        .apply(Create.of(words).withCoder(StringUtf8Coder.of()))%0A        .apply(new WordCount.CountWords())%0A        .apply(MapElements.via(new WordCount.FormatAsTextFn()));%0A%0A    PAssert.that(output).containsInAnyOrder(expectedCounts);%0A%0A    pipeline.run();%0A  %7D%0A%0A  @Test%0A  public void testNamedMetric() throws Exception %7B%0A    assertThat(InMemoryMetrics.valueOf(%22emptyLines%22), is(nullValue()));%0A%0A    runPipeline();%0A%0A    assertThat(InMemoryMetrics.%3CDouble%3EvalueOf(%22emptyLines%22), is(1d));%0A  %7D%0A%7D%0A

Add a test for metrics reporting in Spark runner.