ninadingole

private def buildMovieStream: KStreamS[String, movie.Envelope] = {
    import AppSerdes.movieSerde.consumed
    builder.stream[String, movie.Envelope](Utils.getTopic("movies"))
  }

  private def buildMovieSalesStream = {
    import AppSerdes.movieSalesSerde.consumed
    builder.stream[String, Envelope](Utils.getTopic("sales"))
  }

ninadingole

$>./bin/kafka-topics --zookeeper localhost:2181 --list
__confluent.support.metrics
__consumer_offsets
_confluent-ksql-default__command_topic
_schemas
connect-configs
connect-offsets
connect-status
connect-statuses
dbhistory.inventory

ninadingole

{
  "name": "moviesdemo-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",

ninadingole

mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium' IDENTIFIED BY 'dbz';

ninadingole

[mysqld]
server-id         = 223344   # any random number
log_bin           = mysql-bin
binlog_format     = row      # use row or ROW
binlog_row_image  = full     # use full or FULL
expire_logs_days  = 10
gtid_mode                 = on   # (optional)
enforce_gtid_consistency  = on   # (optional)

ninadingole

{
"class": "io.debezium.connector.mysql.MySqlConnector",
"type": "source",
"version": "0.7.5"
}

ninadingole

> confluent up
Using CONFLUENT_CURRENT: /var/folders/y7/j6zgy2w13g1fjljkgb8mdc4r0000gn/T/confluent.2GgyNcVE
Starting zookeeper
zookeeper is [UP]
Starting Kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting Kafka-rest
kafka-rest is [UP]

ninadingole

CREATE TABLE MOVIES(movie_id INT(10), title VARCHAR(30), year VARCHAR(4), budget INT(8));
CREATE TABLE SALES(movie_id INT(10), sale_id INT(10), total INT(10));

ninadingole

This gist contains all the template I create for my use

ninadingole

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.sample.spring</groupId>
	<artifactId>spring-cookbook</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>pom</packaging>

	<parent>
		<groupId>org.springframework.boot</groupId>