/*
 * 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.
 */
package org.apache.beam.examples.complete.kafkatopubsub;

import static org.apache.beam.examples.complete.kafkatopubsub.kafka.consumer.Utils.configureKafka;
import static org.apache.beam.examples.complete.kafkatopubsub.kafka.consumer.Utils.configureSsl;
import static org.apache.beam.examples.complete.kafkatopubsub.kafka.consumer.Utils.getKafkaCredentialsFromVault;
import static org.apache.beam.examples.complete.kafkatopubsub.kafka.consumer.Utils.isSslSpecified;
import static org.apache.beam.examples.complete.kafkatopubsub.kafka.consumer.Utils.parseKafkaConsumerConfig;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.examples.complete.kafkatopubsub.avro.AvroDataClass;
import org.apache.beam.examples.complete.kafkatopubsub.avro.AvroDataClassKafkaAvroDeserializer;
import org.apache.beam.examples.complete.kafkatopubsub.options.KafkaToPubsubOptions;
import org.apache.beam.examples.complete.kafkatopubsub.transforms.FormatTransform;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link KafkaToPubsub} pipeline is a streaming pipeline which ingests data in JSON format from
 * Kafka, and outputs the resulting records to PubSub. Input topics, output topic, Bootstrap servers
 * are specified by the user as template parameters. <br>
 * Kafka may be configured with SASL/SCRAM security mechanism, in this case a Vault secret storage
 * with credentials should be provided. URL to credentials and Vault token are specified by the user
 * as template parameters.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Kafka Bootstrap Server(s).
 *   <li>Kafka Topic(s) exists.
 *   <li>The PubSub output topic exists.
 *   <li>(Optional) An existing HashiCorp Vault secret storage
 *   <li>(Optional) A configured secure SSL connection for Kafka
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Gradle preparation
 *
 * To run this example your {@code build.gradle} file should contain the following task
 * to execute the pipeline:
 * {@code
 * task execute (type:JavaExec) {
 *     mainClass = System.getProperty("mainClass")
 *     classpath = sourceSets.main.runtimeClasspath
 *     systemProperties System.getProperties()
 *     args System.getProperty("exec.args", "").split()
 * }
 * }
 *
 * This task allows to run the pipeline via the following command:
 * {@code
 * gradle clean execute -DmainClass=org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsub \
 *      -Dexec.args="--<argument>=<value> --<argument>=<value>"
 * }
 *
 * # Running the pipeline
 * To execute this pipeline, specify the parameters:
 *
 * - Kafka Bootstrap servers
 * - Kafka input topics
 * - Pub/Sub output topic
 * - Output format
 *
 * in the following format:
 * {@code
 * --bootstrapServers=host:port \
 * --inputTopics=your-input-topic \
 * --outputTopic=projects/your-project-id/topics/your-topic-pame \
 * --outputFormat=AVRO|PUBSUB
 * }
 *
 * Optionally, to retrieve Kafka credentials for SASL/SCRAM,
 * specify a URL to the credentials in HashiCorp Vault and the vault access token:
 * {@code
 * --secretStoreUrl=http(s)://host:port/path/to/credentials
 * --vaultToken=your-token
 * }
 *
 * Optionally, to configure secure SSL connection between the Beam pipeline and Kafka,
 * specify the parameters:
 * - A path to a truststore file (it can be a local path or a GCS path, which should start with `gs://`)
 * - A path to a keystore file (it can be a local path or a GCS path, which should start with `gs://`)
 * - Truststore password
 * - Keystore password
 * - Key password
 * {@code
 * --truststorePath=path/to/kafka.truststore.jks
 * --keystorePath=path/to/kafka.keystore.jks
 * --truststorePassword=your-truststore-password
 * --keystorePassword=your-keystore-password
 * --keyPassword=your-key-password
 * }
 * By default this will run the pipeline locally with the DirectRunner. To change the runner, specify:
 * {@code
 * --runner=YOUR_SELECTED_RUNNER
 * }
 * </pre>
 *
 * <p><b>Example Avro usage</b>
 *
 * <pre>
 * This template contains an example Class to deserialize AVRO from Kafka and serialize it to AVRO in Pub/Sub.
 *
 * To use this example in the specific case, follow the few steps:
 * <ul>
 * <li> Create your own class to describe AVRO schema. As an example use {@link AvroDataClass}. Just define necessary fields.
 * <li> Create your own Avro Deserializer class. As an example use {@link AvroDataClassKafkaAvroDeserializer}. Just rename it, and put your own Schema class as the necessary types.
 * <li> Modify the {@link FormatTransform}. Put your Schema class and Deserializer to the related parameter.
 * <li> Modify write step in the {@link KafkaToPubsub} by put your Schema class to "writeAvrosToPubSub" step.
 * </ul>
 * </pre>
 */
public class KafkaToPubsub {

  /* Logger for class.*/
  private static final Logger LOG = LoggerFactory.getLogger(KafkaToPubsub.class);

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    KafkaToPubsubOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(KafkaToPubsubOptions.class);

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);
    run(pipeline, options);
  }

  /**
   * Runs a pipeline which reads message from Kafka and writes it to GCS.
   *
   * @param options arguments to the pipeline
   */
  public static PipelineResult run(Pipeline pipeline, KafkaToPubsubOptions options) {
    // Configure Kafka consumer properties
    Map<String, Object> kafkaConfig = new HashMap<>();
    kafkaConfig.putAll(parseKafkaConsumerConfig(options.getKafkaConsumerConfig()));
    Map<String, String> sslConfig = new HashMap<>();
    if (options.getSecretStoreUrl() != null && options.getVaultToken() != null) {
      Map<String, Map<String, String>> credentials =
          getKafkaCredentialsFromVault(options.getSecretStoreUrl(), options.getVaultToken());
      kafkaConfig = configureKafka(credentials.get(KafkaPubsubConstants.KAFKA_CREDENTIALS));
    } else {
      LOG.warn(
          "No information to retrieve Kafka credentials was provided. "
              + "Trying to initiate an unauthorized connection.");
    }

    if (isSslSpecified(options)) {
      sslConfig.putAll(configureSsl(options));
    } else {
      LOG.info(
          "No information to retrieve SSL certificate was provided by parameters."
              + "Trying to initiate a plain text connection.");
    }

    List<String> topicsList = new ArrayList<>(Arrays.asList(options.getInputTopics().split(",")));

    checkArgument(
        topicsList.size() > 0 && topicsList.get(0).length() > 0,
        "inputTopics cannot be an empty string.");

    List<String> bootstrapServersList =
        new ArrayList<>(Arrays.asList(options.getBootstrapServers().split(",")));

    checkArgument(
        bootstrapServersList.size() > 0 && topicsList.get(0).length() > 0,
        "bootstrapServers cannot be an empty string.");

    LOG.info(
        "Starting Kafka-To-PubSub pipeline with parameters bootstrap servers:"
            + options.getBootstrapServers()
            + " input topics: "
            + options.getInputTopics()
            + " output pubsub topic: "
            + options.getOutputTopic());

    /*
     * Steps:
     *  1) Read messages in from Kafka
     *  2) Extract values only
     *  3) Write successful records to PubSub
     */

    if (options.getOutputFormat() == FormatTransform.FORMAT.AVRO) {
      pipeline
          .apply(
              "readAvrosFromKafka",
              FormatTransform.readAvrosFromKafka(
                  options.getBootstrapServers(), topicsList, kafkaConfig, sslConfig))
          .apply("createValues", Values.create())
          .apply("writeAvrosToPubSub", PubsubIO.writeAvros(AvroDataClass.class));

    } else {
      pipeline
          .apply(
              "readFromKafka",
              FormatTransform.readFromKafka(
                  options.getBootstrapServers(), topicsList, kafkaConfig, sslConfig))
          .apply("createValues", Values.create())
          .apply("writeToPubSub", new FormatTransform.FormatOutput(options));
    }

    return pipeline.run();
  }
}
