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

import java.io.File
import scala.collection.Seq
import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits._
import kafka.utils.TestUtils
import org.apache.kafka.common.utils.Time
import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble, MetaPropertiesVersion, PropertiesUtils}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
import org.apache.kafka.test.TestUtils.isValidClusterId

import java.util.Optional


class ServerGenerateClusterIdTest extends QuorumTestHarness {
  var config1: KafkaConfig = _
  var config2: KafkaConfig = _
  var config3: KafkaConfig = _
  var servers: Seq[KafkaServer] = Seq()

  @BeforeEach
  override def setUp(testInfo: TestInfo): Unit = {
    super.setUp(testInfo)
    config1 = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, zkConnect))
    config2 = KafkaConfig.fromProps(TestUtils.createBrokerConfig(2, zkConnect))
    config3 = KafkaConfig.fromProps(TestUtils.createBrokerConfig(3, zkConnect))
  }

  @AfterEach
  override def tearDown(): Unit = {
    TestUtils.shutdownServers(servers)
    super.tearDown()
  }


  @Test
  def testAutoGenerateClusterId(): Unit = {
    // Make sure that the cluster id doesn't exist yet.
    assertFalse(zkClient.getClusterId.isDefined)

    var server1 = createServer(config1, threadNamePrefix = Option(this.getClass.getName))
    servers = Seq(server1)

    // Validate the cluster id
    val clusterIdOnFirstBoot = server1.clusterId
    isValidClusterId(clusterIdOnFirstBoot)

    server1.shutdown()

    // Make sure that the cluster id is persistent.
    assertTrue(zkClient.getClusterId.isDefined)
    assertEquals(zkClient.getClusterId, Some(clusterIdOnFirstBoot))

    // Restart the server check to confirm that it uses the clusterId generated previously
    server1 = createServer(config1, threadNamePrefix = Option(this.getClass.getName))
    servers = Seq(server1)

    val clusterIdOnSecondBoot = server1.clusterId
    assertEquals(clusterIdOnFirstBoot, clusterIdOnSecondBoot)

    server1.shutdown()

    // Make sure that the cluster id is persistent after multiple reboots.
    assertTrue(zkClient.getClusterId.isDefined)
    assertEquals(zkClient.getClusterId, Some(clusterIdOnFirstBoot))

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  @Test
  def testAutoGenerateClusterIdForKafkaClusterSequential(): Unit = {
    val server1 = createServer(config1, threadNamePrefix = Option(this.getClass.getName))
    val clusterIdFromServer1 = server1.clusterId

    val server2 = createServer(config2, threadNamePrefix = Option(this.getClass.getName))
    val clusterIdFromServer2 = server2.clusterId

    val server3 = createServer(config3, threadNamePrefix = Option(this.getClass.getName))
    val clusterIdFromServer3 = server3.clusterId
    servers = Seq(server1, server2, server3)

    servers.foreach(_.shutdown())

    isValidClusterId(clusterIdFromServer1)
    assertEquals(clusterIdFromServer1, clusterIdFromServer2, clusterIdFromServer3)

    // Check again after reboot
    server1.startup()
    assertEquals(clusterIdFromServer1, server1.clusterId)
    server2.startup()
    assertEquals(clusterIdFromServer2, server2.clusterId)
    server3.startup()
    assertEquals(clusterIdFromServer3, server3.clusterId)

    servers.foreach(_.shutdown())

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  @Test
  def testAutoGenerateClusterIdForKafkaClusterParallel(): Unit = {
    val firstBoot = Future.traverse(Seq(config1, config2, config3))(config => Future(createServer(config, threadNamePrefix = Option(this.getClass.getName))))
    servers = Await.result(firstBoot, 100.second)
    val Seq(server1, server2, server3) = servers

    val clusterIdFromServer1 = server1.clusterId
    val clusterIdFromServer2 = server2.clusterId
    val clusterIdFromServer3 = server3.clusterId

    servers.foreach(_.shutdown())
    isValidClusterId(clusterIdFromServer1)
    assertEquals(clusterIdFromServer1, clusterIdFromServer2, clusterIdFromServer3)

    // Check again after reboot
    val secondBoot = Future.traverse(Seq(server1, server2, server3))(server => Future {
      server.startup()
      server
    })
    servers = Await.result(secondBoot, 100.second)
    servers.foreach(server => assertEquals(clusterIdFromServer1, server.clusterId))

    servers.foreach(_.shutdown())

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  @Test
  def testConsistentClusterIdFromZookeeperAndFromMetaProps(): Unit = {
    // Check at the first boot
    val server = createServer(config1, threadNamePrefix = Option(this.getClass.getName))
    val clusterId = server.clusterId

    assertTrue(verifyBrokerMetadata(server.config.logDirs, clusterId))

    server.shutdown()

    // Check again after reboot
    server.startup()

    assertEquals(clusterId, server.clusterId)
    assertTrue(verifyBrokerMetadata(server.config.logDirs, server.clusterId))

    server.shutdown()

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  @Test
  def testInconsistentClusterIdFromZookeeperAndFromMetaProps(): Unit = {
    forgeBrokerMetadata(config1.logDirs, config1.brokerId, "aclusterid")

    val server = new KafkaServer(config1, threadNamePrefix = Option(this.getClass.getName))

    // Startup fails
    assertThrows(classOf[RuntimeException], () => server.startup())

    server.shutdown()

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  @Test
  def testInconsistentBrokerMetadataBetweenMultipleLogDirs(): Unit = {
    // Add multiple logDirs with different BrokerMetadata
    val logDir1 = TestUtils.tempDir().getAbsolutePath
    val logDir2 = TestUtils.tempDir().getAbsolutePath
    val logDirs = logDir1 + "," + logDir2

    forgeBrokerMetadata(logDir1, 1, "ebwOKU-zSieInaFQh_qP4g")
    forgeBrokerMetadata(logDir2, 1, "blaOKU-zSieInaFQh_qP4g")

    val props = TestUtils.createBrokerConfig(1, zkConnect)
    props.setProperty("log.dir", logDirs)
    val config = KafkaConfig.fromProps(props)

    val server = new KafkaServer(config, threadNamePrefix = Option(this.getClass.getName))

    // Startup fails
    assertThrows(classOf[RuntimeException], () => server.startup())

    server.shutdown()

    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
  }

  def forgeBrokerMetadata(logDirs: Seq[String], brokerId: Int, clusterId: String): Unit = {
    for (logDir <- logDirs) {
      forgeBrokerMetadata(logDir, brokerId, clusterId)
    }
  }

  def forgeBrokerMetadata(logDir: String, brokerId: Int, clusterId: String): Unit = {
    val metaProps = new MetaProperties.Builder().
      setVersion(MetaPropertiesVersion.V0).
      setNodeId(brokerId).
      setClusterId(clusterId).
      build()
    PropertiesUtils.writePropertiesFile(metaProps.toProperties,
      new File(logDir, MetaPropertiesEnsemble.META_PROPERTIES_NAME).getAbsolutePath, false)
  }

  def verifyBrokerMetadata(logDirs: Seq[String], clusterId: String): Boolean = {
    for (logDir <- logDirs) {
      val properties = PropertiesUtils.readPropertiesFile(
        new File(logDir, MetaPropertiesEnsemble.META_PROPERTIES_NAME).getAbsolutePath)
      val metaProps = new MetaProperties.Builder(properties).build()
      if (!metaProps.clusterId().equals(Optional.of(clusterId))) {
        return false
      }
    }
    true
  }

  def createServer(config: KafkaConfig, threadNamePrefix: Option[String]): KafkaServer = {
    TestUtils.createServer(config, Time.SYSTEM, threadNamePrefix)
  }
}
