anchor.diff:  
   [source,sql]
   ----
  -2018-06-20 13:03:00 | platinum | 3
  -2018-06-20 13:04:00 | platinum | 26
  -2018-06-20 13:05:00 | platinum | 26
  -2018-06-20 13:06:00 | platinum | 20
  -2018-06-20 13:07:00 | platinum | 23
  +platinum : Window{start=1535986500000 end=-} | platinum | 23
  +silver : Window{start=1535983740000 end=-} | silver | 9
  +gold : Window{start=1535983740000 end=-} | gold | 39
  +gold : Window{start=1535983800000 end=-} | gold | 46
  +platinum : Window{start=1535983980000 end=-} | platinum | 18
   Limit Reached
   Query terminated
   ksql>
  
positive.code:        C.ID, C.CLUB_STATUS, C.EMAIL, \
         C.FIRST_NAME, C.LAST_NAME \
  FROM RATINGS R \
       LEFT JOIN CUSTOMERS C \
         ON R.USER_ID = C.ID \
  WHERE C.FIRST_NAME IS NOT NULL ;
  ----
  
  [source,sql]
  
negative.code: 
  [source,bash]
  ----
  $ docker-compose ps|egrep "elasticsearch|kibana"
  elasticsearch                      /usr/local/bin/docker-entr ...   Up      0.0.0.0:9200->9200/tcp, 0.0.0.0:9300->9300/tcp
  kibana                             /usr/local/bin/kibana-docker     Up      0.0.0.0:5601->5601/tcp
  ----
  
  Create a dynamic mapping in Elasticsearch so that the timestamp of source data is correctly detected:
  
  [source,bash]
  ----
  curl -XPUT "http://localhost:9200/_template/kafkaconnect/" -H 'Content-Type: application/json' -d' { "index_patterns": "*", "settings": { "number_of_shards": 1, "number_of_replicas": 0 }, "mappings": { "_default_": { "dynamic_templates": [ { "dates": { "match": "TS", "mapping": { "type": "date" } } }, { "non_analysed_string_template": { "match": "*", "match_mapping_type": "string", "mapping": { "type": "keyword" } } } ] } } }'
  ----
  
  Create a connector to stream `RATINGS_WITH_CUSTOMER_DATA` to Elasticsearch:
  