/*
 * 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.
 */

/**
 * Main Flink Runner build file shared by all of its build targets.
 * The file needs to be parameterized by the Flink version and the source directories.
 *
 * See build.gradle files for an example of how to use this script.
 */

import groovy.json.JsonOutput

apply plugin: 'org.apache.beam.module'
applyJavaNature(
    enableStrictDependencies:true,
    automaticModuleName: 'org.apache.beam.runners.flink',
    archivesBaseName: (project.hasProperty('archives_base_name') ? archives_base_name : archivesBaseName),
)

description = "Apache Beam :: Runners :: Flink $flink_version"

/*
 * 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(":sdks:java:core")
evaluationDependsOn(":runners:core-java")

/*
 * Copy & merge source overrides into build directory.
 */
def sourceOverridesBase = "${project.buildDir}/source-overrides/src"

def copySourceOverrides = tasks.register('copySourceOverrides', Copy) {
  it.from main_source_overrides
  it.into "${sourceOverridesBase}/main/java"
  it.duplicatesStrategy DuplicatesStrategy.INCLUDE
}
compileJava.dependsOn copySourceOverrides

def copyResourcesOverrides = tasks.register('copyResourcesOverrides', Copy) {
  it.from main_resources_overrides
  it.into "${sourceOverridesBase}/main/resources"
  it.duplicatesStrategy DuplicatesStrategy.INCLUDE
}
compileJava.dependsOn copyResourcesOverrides

def copyTestSourceOverrides = tasks.register('copyTestSourceOverrides', Copy) {
  it.from test_source_overrides
  it.into "${sourceOverridesBase}/test/java"
  it.duplicatesStrategy DuplicatesStrategy.INCLUDE
}
compileTestJava.dependsOn copyTestSourceOverrides

def copyTestResourcesOverrides = tasks.register('copyTestResourcesOverrides', Copy) {
  it.from test_resources_overrides
  it.into "${sourceOverridesBase}/test/resources"
  it.duplicatesStrategy DuplicatesStrategy.INCLUDE
}
compileJava.dependsOn copyTestResourcesOverrides

/*
 * We have to explicitly set all directories here to make sure each
 * version of Flink has the correct overrides set.
 */
def sourceBase = "${project.projectDir}/../src"
sourceSets {
  main {
    java {
      srcDirs = ["${sourceBase}/main/java", "${sourceOverridesBase}/main/java"]
    }
    resources {
      srcDirs = ["${sourceBase}/main/resources", "${sourceOverridesBase}/main/resources"]
    }
  }
  test {
    java {
      srcDirs = ["${sourceBase}/test/java", "${sourceOverridesBase}/test/java"]
    }
    resources {
      srcDirs = ["${sourceBase}/test/resources", "${sourceOverridesBase}/test/resources"]
    }
  }
}

test {
  systemProperty "log4j.configuration", "log4j-test.properties"
  // Change log level to debug:
  // systemProperty "org.slf4j.simpleLogger.defaultLogLevel", "debug"
  // Change log level to debug only for the package and nested packages:
  // systemProperty "org.slf4j.simpleLogger.log.org.apache.beam.runners.flink.translation.wrappers.streaming", "debug"
  jvmArgs "-XX:-UseGCOverheadLimit"
  if (System.getProperty("beamSurefireArgline")) {
    jvmArgs System.getProperty("beamSurefireArgline")
  }
  // TODO(BEAM-6418) Running tests of all Flink versions in parallel can be too harsh on Jenkins memory.
  // Run them serially for now, to avoid "Exit code 137", i.e. Jenkins host killing the Gradle test process.
  def flink_minor_version = project.path.split(':').last()
  for (version in project.ext.allFlinkVersions) {
    if (version == flink_minor_version) {
      break
    }
    mustRunAfter(":runners:flink:${version}:test")
  }
}

configurations {
  validatesRunner
  miniCluster
}

dependencies {
  compileOnly project(":sdks:java:build-tools")
  compile library.java.vendored_guava_26_0_jre
  compile project(path: ":sdks:java:core", configuration: "shadow")
  compile project(":runners:core-java")
  compile project(":runners:core-construction-java")
  compile project(":runners:java-fn-execution")
  compile project(":runners:java-job-service")
  compile project(":sdks:java:extensions:google-cloud-platform-core")
  compile library.java.vendored_grpc_1_26_0
  compile library.java.slf4j_api
  compile library.java.joda_time
  compile library.java.args4j
  compile "org.apache.flink:flink-clients_2.11:$flink_version"
  // Runtime dependencies are not included in Beam's generated pom.xml, so we must declare flink-clients in compile
  // configuration (https://issues.apache.org/jira/browse/BEAM-11732).
  permitUnusedDeclared "org.apache.flink:flink-clients_2.11:$flink_version"
  compile "org.apache.flink:flink-core:$flink_version"
  compile "org.apache.flink:flink-metrics-core:$flink_version"
  compile "org.apache.flink:flink-java:$flink_version"
  compile "org.apache.flink:flink-runtime_2.11:$flink_version"
  compile "org.apache.flink:flink-streaming-java_2.11:$flink_version"
  // RocksDB state backend (included in the Flink distribution)
  provided "org.apache.flink:flink-statebackend-rocksdb_2.11:$flink_version"
  testCompile project(path: ":sdks:java:core", configuration: "shadowTest")
  // FlinkStateInternalsTest extends abstract StateInternalsTest
  testCompile project(path: ":runners:core-java", configuration: "testRuntime")
  testCompile library.java.hamcrest_core
  testCompile library.java.junit
  testCompile library.java.mockito_core
  testCompile library.java.powermock
  testCompile library.java.google_api_services_bigquery
  testCompile project(":sdks:java:io:google-cloud-platform")
  testCompile library.java.jackson_dataformat_yaml
  testCompile "org.apache.flink:flink-core:$flink_version:tests"
  testCompile "org.apache.flink:flink-runtime_2.11:$flink_version:tests"
  testCompile "org.apache.flink:flink-streaming-java_2.11:$flink_version:tests"
  testCompile "org.apache.flink:flink-test-utils_2.11:$flink_version"
  testCompile project(":sdks:java:harness")
  testRuntimeOnly library.java.slf4j_simple
  validatesRunner project(path: ":sdks:java:core", configuration: "shadowTest")
  validatesRunner project(path: ":runners:core-java", configuration: "testRuntime")
  validatesRunner project(project.path)
  miniCluster "org.apache.flink:flink-runtime-web_2.11:$flink_version"
  compile project(path: ":model:fn-execution", configuration: "shadow")
  compile project(path: ":model:pipeline", configuration: "shadow")
  compile project(path: ":model:job-management", configuration: "shadow")
  compile project(":sdks:java:fn-execution")
  compile library.java.jackson_databind
  compile "org.apache.flink:flink-annotations:$flink_version"
  compile "org.apache.flink:flink-optimizer_2.11:$flink_version"
}

class ValidatesRunnerConfig {
  String name
  boolean streaming
  boolean checkpointing
}

def createValidatesRunnerTask(Map m) {
  def config = m as ValidatesRunnerConfig
  tasks.register(config.name, Test) {
    group = "Verification"
    def runnerType = config.streaming ? "streaming" : "batch"
    description = "Validates the ${runnerType} runner"
    def pipelineOptionsArray = ["--runner=TestFlinkRunner",
                                "--streaming=${config.streaming}",
                                "--parallelism=2",
    ]
    if (config.checkpointing) {
      pipelineOptionsArray.addAll([
              "--checkpointingInterval=3000",
              "--shutdownSourcesAfterIdleMs=60000",
      ])
    }
    def pipelineOptions = JsonOutput.toJson(pipelineOptionsArray)
    systemProperty "beamTestPipelineOptions", pipelineOptions
    classpath = configurations.validatesRunner
    testClassesDirs = files(
      project(":sdks:java:core").sourceSets.test.output.classesDirs,
      project(":runners:core-java").sourceSets.test.output.classesDirs,
    )
    // maxParallelForks decreased from 4 in order to avoid OOM errors
    maxParallelForks 2
    useJUnit {
      if (config.checkpointing) {
        includeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
        // TestStreamSource does not support checkpointing
        excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
      } else {
        includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
        excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
      }
      excludeCategories 'org.apache.beam.sdk.testing.FlattenWithHeterogeneousCoders'
      excludeCategories 'org.apache.beam.sdk.testing.LargeKeys$Above100MB'
      excludeCategories 'org.apache.beam.sdk.testing.UsesCommittedMetrics'
      excludeCategories 'org.apache.beam.sdk.testing.UsesSystemMetrics'
      excludeCategories 'org.apache.beam.sdk.testing.UsesOnWindowExpiration'
      excludeCategories 'org.apache.beam.sdk.testing.UsesStrictTimerOrdering'
      excludeCategories 'org.apache.beam.sdk.testing.UsesOrderedListState'
      if (config.streaming) {
        excludeCategories 'org.apache.beam.sdk.testing.UsesTimerMap'
        excludeCategories 'org.apache.beam.sdk.testing.UsesTestStreamWithMultipleStages'  // BEAM-8598
        excludeCategories 'org.apache.beam.sdk.testing.UsesTestStreamWithProcessingTime'
      } else {
        excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
        excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
        excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo'
        excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
      }
      filter {
        // https://issues.apache.org/jira/browse/BEAM-12039
        excludeTestsMatching 'org.apache.beam.sdk.testing.TestStreamTest.testDiscardingMode'
        // https://issues.apache.org/jira/browse/BEAM-12037
        excludeTestsMatching 'org.apache.beam.sdk.testing.TestStreamTest.testFirstElementLate'
        // https://issues.apache.org/jira/browse/BEAM-12038
        excludeTestsMatching 'org.apache.beam.sdk.testing.TestStreamTest.testLateDataAccumulating'
      }
    }
  }
}

createValidatesRunnerTask(name: "validatesRunnerBatch", streaming: false)
createValidatesRunnerTask(name: "validatesRunnerStreaming", streaming: true)
// We specifically have a variant which runs with checkpointing enabled for the
// tests that require it since running a checkpoint variant is significantly
// slower since we have to wait shutdownSourcesAfterIdleMs before the source
// can shutdown because of https://issues.apache.org/jira/browse/FLINK-2491
// not supporting checkpointing when an operator has been shutdown.
createValidatesRunnerTask(name: "validatesRunnerStreamingCheckpointing", streaming: true, checkpointing: true)

tasks.register('validatesRunner') {
  group = 'Verification'
  description "Validates Flink runner"
  dependsOn validatesRunnerBatch
  dependsOn validatesRunnerStreaming
  dependsOn validatesRunnerStreamingCheckpointing
}

// Generates :runners:flink:1.10:runQuickstartJavaFlinkLocal
createJavaExamplesArchetypeValidationTask(type: 'Quickstart', runner: 'FlinkLocal')

/**
 * Updates the documentation with the current pipeline options.
 */
def createPipelineOptionsTableTask(String target) {
  tasks.register("generatePipelineOptionsTable${target}", JavaExec) {
    group = 'Website'
    description = "Generates a table with pipeline options for the Flink Runner documentation page"
    classpath = sourceSets.test.runtimeClasspath
    main = 'org.apache.beam.runners.flink.website.PipelineOptionsTableGenerator'
    args = [target]
    standardOutput = new ByteArrayOutputStream()
    doLast {
      def dest = file("${project(':website').getProjectDir()}/www/site/layouts/shortcodes/flink_${target.toLowerCase()}_pipeline_options.html")
      if (!dest.exists()) {
        throw new GradleException("Pipeline options file is not in expected location: ${dest}")
      }
      dest.write(standardOutput.toString())
    }
  }
}
createPipelineOptionsTableTask('Java')
createPipelineOptionsTableTask('Python')
// Update the pipeline options documentation before running the tests
test.dependsOn(generatePipelineOptionsTableJava)
test.dependsOn(generatePipelineOptionsTablePython)