docker-compose.yml

version: "3"

networks:
  kafka_workshop:
    driver: bridge

services:
  zeppelin:
    hostname: zeppelin
    container_name: zeppelin
    image: larusefraudy/zeppelin:0.9.0
    depends_on:
      - neo4j
    ports:
      - "8080:8080"
      - "4040:4040"
    volumes:
      - ./zeppelin/notebook:/zeppelin/notebook
      - ./zeppelin/conf:/zeppelin/conf
      - ./zeppelin/interpreter/neo4j:/zeppelin/interpreter/neo4j
    networks:
      - kafka_workshop

  neo4j:
    platform: linux/amd64
    image: neo4j:4.3-enterprise
    hostname: neo4j
    container_name: neo4j
    ports:
      - 7474:7474
      - 7687:7687
    depends_on:
      - zookeeper
      - broker
    volumes:
      - ./neo4j/plugins:/plugins
    environment:
      NEO4J_ACCEPT_LICENSE_AGREEMENT: "yes"
      NEO4J_AUTH: neo4j/password
      NEO4J_dbms_memory_heap_max__size: 2G
      NEO4J_dbms_logs_debug_level: DEBUG
      NEO4J_apoc_export_file_enabled: "true"
      NEO4J_kafka_zookeeper_connect: zookeeper:2181
      NEO4J_kafka_bootstrap_servers: broker:9093
      NEO4J_kafka_client_id: "neo4jClient"
      NEO4J_kafka_group_id: "neo4jGroup"
      NEO4J_kafka_key_deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      NEO4J_kafka_value_deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      NEO4J_streams_source_enabled: "true"
      NEO4J_streams_sink_enabled: "true"
      NEO4J_streams_source_topic_nodes_customer: Customer{customerID,contactName,companyName}
      NEO4J_streams_source_topic_nodes_order: Order{orderID,shipCity,shipAddress}
      NEO4J_streams_source_topic_nodes_product: Product{productID,productName}
      NEO4J_streams_source_topic_nodes_people: Person{*}
      NEO4J_streams_source_topic_relationships_knows: KNOWS{*}
      NEO4J_streams_source_topic_relationships_purchased: PURCHASED{*}
      NEO4J_streams_source_topic_relationships_orders: ORDERS{quantity,unitPrice}
      NEO4J_streams_sink_topic_cypher_sales: "
            MERGE (c:Customer {customerID: event.customer.customerID})
            ON CREATE SET c.contactName = event.customer.contactName,
            c.companyName = event.customer.companyName
            MERGE (o:Order {orderID: event.order.orderID})
            ON CREATE SET o.shipCity = event.order.shipCity,
            o.shipAddress = event.order.shipAddress,
            o.orderDate = localdatetime()
            MERGE (p:Product {productID: event.product.productID})
            ON CREATE SET p.productName = event.product.productName
            MERGE (c)-[:PURCHASED]->(o)-[os:ORDERS {quantity: event.product.quantity, unitPrice: event.product.unitPrice}]->(p)"
    networks:
      - kafka_workshop

  zookeeper:
    image: confluentinc/cp-zookeeper
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - kafka_workshop

  broker:
    image: confluentinc/cp-enterprise-kafka
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    expose:
      - "9093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9093,OUTSIDE://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9093,OUTSIDE://0.0.0.0:9092
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:9093
      # workaround if we change to a custom name the schema_registry fails to start
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: "true"
      CONFLUENT_SUPPORT_CUSTOMER_ID: "anonymous"
    networks:
      - kafka_workshop

  connect:
    image: confluentinc/cp-kafka-connect
    hostname: connect
    container_name: connect
    depends_on:
      - zookeeper
      - broker
    ports:
      - "8083:8083"
    volumes:
      - ./kafka-connect/plugins:/tmp/connect-plugins
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "broker:9093"
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_ZOOKEEPER_CONNECT: "zookeeper:2181"
      CONNECT_PLUGIN_PATH: /usr/share/java,/tmp/connect-plugins,/usr/share/confluent-hub-components
      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=DEBUG,org.I0Itec.zkclient=DEBUG,org.reflections=ERROR,org.apache.kafka.connect.transforms
    networks:
      - kafka_workshop