1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.docker
21 
22 import com.whisk.docker.{ContainerLink, DockerContainer, DockerReadyChecker}
23 import org.scalatest.Suite
24 
25 trait DockerKafkaService extends DockerZookeeperService {
26   self: Suite =>
27 
28   val kafkaInternalPort: Int = 9092
29   private val defaultKafkaExposedPort: Int = 29092
30   def kafkaExposedPort: Int = defaultKafkaExposedPort
31   def kafkaName: String = "kafka"
32   def brokerId: Int = 1
33 
34   val registryInternalPort: Int = 8081
35   private val defaultRegistryExposedPort: Int = 28081
36   def registryExposedPort: Int = defaultRegistryExposedPort
37   def registryName: String = "registry"
38 
39   protected def kafkaContainer: DockerContainer =
40     DockerContainer(s"confluentinc/cp-kafka:${ConfluentPlatformTest.confluentVersion}", name = Some(kafkaName))
41       .withEnv(
42         s"KAFKA_ZOOKEEPER_CONNECT=$zookeeperName:$zookeeperInternalPort/kafka",
43         s"KAFKA_ADVERTISED_LISTENERS=EXTERNAL://127.0.0.1:$kafkaExposedPort,INTERNAL://$kafkaName:$kafkaInternalPort",
44         s"KAFKA_LISTENERS=INTERNAL://0.0.0.0:$kafkaInternalPort,EXTERNAL://0.0.0.0:$kafkaExposedPort",
45         "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT",
46         s"KAFKA_BROKER_ID=$brokerId",
47         "KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL",
48         "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1"
49       )
50       .withHostname(kafkaName)
51       .withLinks(ContainerLink(zookeeperContainer, zookeeperName))
52       .withPorts(kafkaExposedPort -> Some(kafkaExposedPort))
53       .withReadyChecker(DockerReadyChecker.LogLineContains("started (kafka.server.KafkaServer)"))
54 
55   protected def avroRegistryContainer: DockerContainer =
56     DockerContainer(
57       s"confluentinc/cp-schema-registry:${ConfluentPlatformTest.confluentVersion}",
58       name = Some(registryName)
59     ).withEnv(
60         s"SCHEMA_REGISTRY_HOST_NAME=$registryName",
61         s"SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=PLAINTEXT://$kafkaName:$kafkaInternalPort",
62         "SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL=PLAINTEXT",
63         s"SCHEMA_REGISTRY_LISTENERS=http://$registryName:$registryInternalPort"
64       )
65       .withHostname(registryName)
66       .withPorts(registryInternalPort -> Some(registryExposedPort))
67       .withLinks(ContainerLink(kafkaContainer, kafkaName))
68       .withReadyChecker(DockerReadyChecker.LogLineContains("Server started, listening for requests..."))
69 
70   abstract override def dockerContainers: List[DockerContainer] =
71     avroRegistryContainer :: kafkaContainer :: super.dockerContainers
72 }