build.gradle

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * License); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an AS IS BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import groovy.json.JsonOutput
plugins { id 'org.apache.beam.module' }

// Shade away runner execution utilities till because this causes ServiceLoader conflicts with
// TransformPayloadTranslatorRegistrar amongst other runners. This only happens in the DirectRunner
// because it is likely to appear on the classpath of another runner.
def dependOnProjects = [":runners:core-construction-java",
                        ":runners:core-java",
                        ":runners:local-java",
                        ":runners:java-fn-execution",
                        ":sdks:java:fn-execution",
                        ":sdks:java:extensions:avro"
                        ]

applyJavaNature(
        automaticModuleName: 'org.apache.beam.runners.direct',
        classesTriggerCheckerBugs: [
          'ImpulseEvaluatorFactory': 'https://github.com/typetools/checker-framework/issues/3791',
        ],
        shadowClosure: {
          dependencies {
            dependOnProjects.each {
              include(project(path: it, configuration: "shadow"))
            }
          }
          relocate "org.apache.beam.runners.core", getJavaRelocatedPath("runners.core")
          relocate "org.apache.beam.runners.fnexecution", getJavaRelocatedPath("runners.fnexecution")
          relocate "org.apache.beam.sdk.fn", getJavaRelocatedPath("sdk.fn")
          relocate "org.apache.beam.runners.local", getJavaRelocatedPath("runners.local")
        },
)

description = "Apache Beam :: Runners :: Direct Java"

/*
 * We need to rely on manually specifying these evaluationDependsOn to ensure that
 * the following projects are evaluated before we evaluate this project. This is because
 * we are attempting to reference the "sourceSets.test.output" directly.
 */
evaluationDependsOn(":runners:core-construction-java")
evaluationDependsOn(":runners:core-java")
evaluationDependsOn(":sdks:java:core")
evaluationDependsOn(":examples:java")

configurations {
  needsRunner
  validatesRunner
  examplesJavaIntegrationTest
}

dependencies {
  shadow library.java.vendored_guava_26_0_jre
  shadow project(path: ":model:pipeline", configuration: "shadow")
  dependOnProjects.each {
    implementation project(it)
  }
  shadow project(path: ":sdks:java:core", configuration: "shadow")
  shadow library.java.vendored_grpc_1_48_1
  shadow library.java.joda_time
  shadow library.java.slf4j_api
  shadow library.java.jackson_databind
  provided library.java.hamcrest
  permitUnusedDeclared library.java.hamcrest
  provided library.java.junit
  permitUnusedDeclared library.java.junit
  shadowTest project(path: ":sdks:java:core", configuration: "shadowTest")
  shadowTest project(path: ":runners:core-java", configuration: "testRuntimeMigration")
  shadowTest library.java.mockito_core
  needsRunner project(path: ":runners:core-construction-java", configuration: "testRuntimeMigration")
  needsRunner project(path: ":runners:core-java", configuration: "testRuntimeMigration")
  needsRunner project(path: ":sdks:java:core", configuration: "shadowTest")
  needsRunner project(path: project.path, configuration: "shadow")
  needsRunner project(path: project.path, configuration: "shadowTest")
  validatesRunner project(path: ":runners:core-construction-java", configuration: "testRuntimeMigration")
  validatesRunner project(path: ":runners:core-java", configuration: "testRuntimeMigration")
  validatesRunner project(path: ":sdks:java:core", configuration: "shadowTest")
  validatesRunner project(path: project.path, configuration: "shadow")
  validatesRunner project(path: project.path, configuration: "shadowTest")
  permitUnusedDeclared library.java.vendored_grpc_1_48_1
  permitUnusedDeclared project(":runners:java-fn-execution")
  permitUnusedDeclared project(":sdks:java:fn-execution")
  permitUnusedDeclared project(":sdks:java:extensions:avro")
  examplesJavaIntegrationTest project(project.path)
  examplesJavaIntegrationTest project(":examples:java")
  examplesJavaIntegrationTest project(path: ":examples:java", configuration: "testRuntimeMigration")
}

// windows handles quotes differently from linux,
// on linux, the pipeline option string looks like
// "--runner=DirectRunner", "--runnerDeterminedSharding=false"
// on windows, the pipeline option string needs to be
// "\"--runner=DirectRunner\"", "\"--runnerDeterminedSharding=false\""
// to achieve the same effect
static def pipelineOptionsStringCrossPlatformHandling(String[] options) {
  if (System.properties['os.name'].toLowerCase().contains('windows')) {
    return JsonOutput.toJson(options.collect { "\"$it\"" })
  } else {
    return JsonOutput.toJson(options)
  }
}

def sickbayTests = [
        // https://github.com/apache/beam/issues/18499
        'org.apache.beam.sdk.testing.UsesLoopingTimer',
        // https://github.com/apache/beam/issues/19344
        'org.apache.beam.sdk.io.BoundedReadFromUnboundedSourceTest.testTimeBound',
]

task needsRunnerTests(type: Test) {
  group = "Verification"
  description = "Runs tests that require a runner to validate that piplines/transforms work correctly"

  testLogging.showStandardStreams = true

  String[] pipelineOptions = ["--runner=DirectRunner", "--runnerDeterminedSharding=false"]
  systemProperty "beamTestPipelineOptions", pipelineOptionsStringCrossPlatformHandling(pipelineOptions)

  classpath = configurations.needsRunner
  // NOTE: We only add to the test class dirs to ensure that the direct runner
  // tests (i.e., those from this subproject) get scanned.
  testClassesDirs += files(project(":runners:core-construction-java").sourceSets.test.output.classesDirs)
  testClassesDirs += files(project(":runners:core-java").sourceSets.test.output.classesDirs)
  testClassesDirs += files(project(":sdks:java:core").sourceSets.test.output.classesDirs)
  useJUnit {

    filter {
      for (String test : sickbayTests) {
        excludeTestsMatching test
      }
    }

    includeCategories "org.apache.beam.sdk.testing.NeedsRunner"
    // Should be run only in a properly configured SDK harness environment
    excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment'
    excludeCategories "org.apache.beam.sdk.testing.LargeKeys\$Above100MB"
    // MetricsPusher isn't implemented in direct runner
    excludeCategories "org.apache.beam.sdk.testing.UsesMetricsPusher"
    excludeCategories "org.apache.beam.sdk.testing.UsesJavaExpansionService"
    excludeCategories "org.apache.beam.sdk.testing.UsesPythonExpansionService"
    excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
  }
  testLogging {
    outputs.upToDateWhen {false}
    showStandardStreams = true
  }
}

// NOTE: This will also run 'NeedsRunner' tests, which are run in the :needsRunnerTests task as well.
// The intention of this task is to mirror the :validatesRunner configuration for other runners,
// such that the test suite can be validated on the in-process DirectRunner.
task validatesRunner(type: Test) {
  group = "Verification"
  description "Validates Direct runner"

  String[] pipelineOptions = ["--runner=DirectRunner", "--runnerDeterminedSharding=false"]
  systemProperty "beamTestPipelineOptions", pipelineOptionsStringCrossPlatformHandling(pipelineOptions)

  classpath = configurations.validatesRunner
  // NOTE: We only add to the test class dirs to ensure that the direct runner
  // tests (i.e., those from this subproject) get scanned.
  testClassesDirs += files(project(":runners:core-construction-java").sourceSets.test.output.classesDirs)
  testClassesDirs += files(project(":runners:core-java").sourceSets.test.output.classesDirs)
  testClassesDirs += files(project(":sdks:java:core").sourceSets.test.output.classesDirs)
  useJUnit {
    includeCategories "org.apache.beam.sdk.testing.ValidatesRunner"
    // Should be run only in a properly configured SDK harness environment
    excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment'
    excludeCategories "org.apache.beam.sdk.testing.LargeKeys\$Above100MB"
    excludeCategories 'org.apache.beam.sdk.testing.UsesMetricsPusher'
    excludeCategories "org.apache.beam.sdk.testing.UsesJavaExpansionService"
    excludeCategories "org.apache.beam.sdk.testing.UsesPythonExpansionService"
    // https://github.com/apache/beam/issues/18499
    excludeCategories 'org.apache.beam.sdk.testing.UsesLoopingTimer'
  }
}

tasks.register("validatesRunnerSickbay", Test) {
  group = "Verification"
  description "Validates Direct runner (Sickbay Tests)"
  systemProperty "beamTestPipelineOptions", JsonOutput.toJson([
          "--runner=DirectRunner",
  ])

  classpath = configurations.needsRunner
  testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs)

  filter {
    for (String test : sickbayTests) {
      includeTestsMatching test
    }

    // https://github.com/apache/beam/issues/18826
    includeTestsMatching 'org.apache.beam.runners.direct.WatermarkManagerTest.updateWatermarkWithDifferentWindowedValueInstances'
  }
}

def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
def gcsBucket = project.findProperty('gcsBucket') ?: 'temp-storage-for-release-validation-tests/nightly-snapshot-validation'
def bqDataset = project.findProperty('bqDataset') ?: 'beam_postrelease_mobile_gaming'
def pubsubTopic = project.findProperty('pubsubTopic') ?: 'java_mobile_gaming_topic'
def tempLocation = project.findProperty('tempLocation') ?: 'gs://temp-storage-for-end-to-end-tests'

// Generates :runners:direct-java:runQuickstartJavaDirect
createJavaExamplesArchetypeValidationTask(type: 'Quickstart', runner: 'Direct')

// Generates :runners:direct-java:runMobileGamingJavaDirect
createJavaExamplesArchetypeValidationTask(type: 'MobileGaming',
  runner: 'Direct',
  gcpProject: gcpProject,
  gcsBucket: gcsBucket,
  bqDataset: bqDataset,
  pubsubTopic: pubsubTopic)

task examplesIntegrationTest(type: Test) {
  description = "Runs examples tests on DirectRunner"

  testLogging.showStandardStreams = true

  String[] pipelineOptions = [
          "--runner=DirectRunner",
          "--runnerDeterminedSharding=false",
          "--tempLocation=${tempLocation}",
          "--tempRoot=${tempLocation}",
          "--project=${gcpProject}",
  ]
  systemProperty "beamTestPipelineOptions", pipelineOptionsStringCrossPlatformHandling(pipelineOptions)

  include '**/*IT.class'
  maxParallelForks 4
  classpath = configurations.examplesJavaIntegrationTest
  testClassesDirs = files(project(":examples:java").sourceSets.test.output.classesDirs)
  useJUnit {
    filter{
      // TODO (https://github.com/apache/beam/issues/21344) Fix integration Tests to run with DirectRunner: Timeout error
      excludeTestsMatching 'org.apache.beam.examples.complete.TfIdfIT'
      excludeTestsMatching 'org.apache.beam.examples.WindowedWordCountIT.testWindowedWordCountInBatchDynamicSharding'
    }
  }
}