title: Unable to read messages from Kafka topic using console consumer
body: <p>I created <code>stream1</code> in <code>KSQL</code> (version 5.0 Beta) with a backing topic <code>topic1</code> and <code>avro</code> schema. I am able to read all messages on <code>topic1</code> using <code>kafka-avro-console-consumer</code>.  </p>
  
  <p>I then created <code>stream2</code> in <code>KSQL</code> that's based on <code>stream1</code> but with <code>json</code> format for the messages and backing topic named <code>topic2</code>. I am able to read all the messages on <code>topic2</code> using <code>kafka-console-consumer</code></p>
  
  <p>I created <code>stream3</code> in <code>KSQL</code> based on <code>stream2</code> with <code>json</code> message format and backing topic named <code>topic3</code>. However, I am unable to read the messages on <code>topic3</code> using <code>kafka-console-consumer</code>. </p>
  
  <p>Using <code>kafkacat</code> I get offsets on various partitions on <code>topic3</code> but none of the actual messages is being printed.</p>
  
  <p>It looks likes the messages are in the topic but neither <code>kafkacat</code> not <code>kafka-console-consumer</code> is able to print it. </p>
  
  <p>Tried using <code>--from-beginning</code> and <code>--offset earliest --partition 0</code> with no luck.  </p>
  
  <p>Here are the KSQL statements</p>
  
  <pre><code>CREATE STREAM stream1(p_id STRING, location STRING, u_id STRING, r_id STRING, b_id STRING, recorded_dtm STRING, 
  v_type STRING, value STRING) WITH (kafka_topic='topic1', value_format='AVRO');
  
  CREATE STREAM stream2 WITH (KAFKA_topic='topic2', VALUE_FORMAT='json', TIMESTAMP='RECORDED_TIMESTAMP') 
  AS select P_ID+'-'+LOCATION+'-'+U_ID+'-'+R_ID+'-'+B_ID+'-'+V_TYPE as PARTITION_KEY, 
  LOCATION, U_ID, R_ID, V_TYPE, B_ID, STRINGTOTIMESTAMP(recorded_dtm, 'yyyyMMddHHmmss') as RECORDED_TIMESTAMP, 
  P_ID, VALUE, RECORDED_DTM,'NM' as DATA_TYPE 
  FROM stream1 PARTITION BY PARTITION_KEY;
  
  CREATE STREAM stream3 WITH (KAFKA_topic='topic3', VALUE_FORMAT='json', TIMESTAMP='RECORDED_TIMESTAMP') 
  AS select PARTITION_KEY, LOCATION, U_ID, R_ID, V_TYPE, B_ID, RECORDED_TIMESTAMP, 
  P_ID, VALUE, RECORDED_DTM FROM stream2 PARTITION BY PARTITION_KEY;
  </code></pre>
  
  <p>Additional info</p>
  
  <p>In <code>ksql</code> if I run <code>SET 'auto.offset.reset'='earliest';</code> and run <code>select * from stream1 limit 5;</code> or <code>select * from stream2 limit 5</code> I see records printed but <code>select * from stream3 limit 5</code> does not return any records.</p>
  
  <p>If I run <code>describe extended stream3</code> I get </p>
  
  <p>total-messages: 212 </p>
  
  <p>which happens to be the number of messages I sent to topic1</p>