the right way to see changelog persistent by rocksdb is use ByteDeser, and
then decode hex to string
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.ByteArrayDeserializer");
for(ConsumerRecord<String, byte[]> record: consumerRecords) {
print bytesToHexString(record.value())
}
public static String bytesToHexString(byte[] src){
StringBuilder stringBuilder = new StringBuilder("");
if (src == null || src.length <= 0) {
return null;
}
for (int i = 0; i < src.length; i++) {
int v = src[i] & 0xFF;
String hv = Integer.toHexString(v);
if (hv.length() < 2) {
stringBuilder.append(0);
}
stringBuilder.append(hv);
}
return stringBuilder.toString();
}
output now collect
p:0,o:0,k:msg1,v:00000001
p:0,o:1,k:msg3,v:00000001
p:0,o:2,k:msg5,v:00000001
p:0,o:3,k:msg1,v:00000002
p:0,o:4,k:msg3,v:00000002
p:1,o:0,k:msg2,v:00000001
p:1,o:1,k:msg4,v:00000001
p:1,o:2,k:msg2,v:00000002
p:1,o:3,k:msg2,v:00000003
2017-06-07 18:42 GMT+08:00 john cheng <zqhxuy...@gmail.com>:
> I add some log on StoreChangeLog
>
> for (K k : this.dirty) {
> V v = getter.get(k);
> log.info("logChange key:{},value:{}", k, v);
> collector.send(new ProducerRecord<>(this.topic, this.partition, k, v),
> keySerializer, valueSerializer);
> }
>
> and found the print result is normal, just some byte:
>
> [2017-06-07 18:39:43,131] INFO logChange removed:[], dirty:[kafka]
> (org.apache.kafka.streams.state.internals.StoreChangeLogger)
> [2017-06-07 18:39:43,132] INFO logChange key:kafka,value:[0, 0, 0, 2]
> (org.apache.kafka.streams.state.internals.StoreChangeLogger)
>
> but console-consumer still output kafka:
>
> so this may be problem of console-consume.
>
>
> 2017-06-07 18:24 GMT+08:00 john cheng <zqhxuy...@gmail.com>:
>
>> I'm running WordCountProcessorDemo with Processor API. and change
>> something below
>> 1. config 1 stream-thread and 1 replicas
>> 2. change inMemory() to persistent()
>> MyKakfa version is 0.10.0.0. After running streaming application, I check
>> msg output by console-consumer
>> ➜ kafka_2.10-0.10.0.0 bin/kafka-console-producer.sh --broker-list
>> localhost:9092 --topic streams-input2
>> msg1 # by console-producer, we can only produce message's value. so
>> message produce to input topic will use roundrobbin partition
>> msg2
>> msg3
>> msg4
>> msg5
>> msg6
>> ➜ kafka_2.10-0.10.0.0 bin/kafka-console-consumer.sh --new-consumer
>> --bootstrap-server localhost:9092 --topic streams-input2 --property
>> print.key=true --property key.separator=":" --from-beginning
>> null:msg2 # key is null, value is what we produce above
>> null:msg4
>> null:msg6
>> null:msg1
>> null:msg3
>> null:msg5
>> ➜ kafka_2.10-0.10.0.0 bin/kafka-console-consumer.sh --new-consumer
>> --bootstrap-server localhost:9092 --topic streams-output2 --property
>> print.key=true --property key.separator=":" --from-beginning
>> msg2:1
>> msg1:1
>> msg1:1
>> msg3:1
>> msg2:1
>> msg4:1
>> msg1:1
>> msg3:1
>> msg5:1
>> msg2:1
>> msg4:1
>> msg6:1 # due to log compaction, same key will be overwrite. this is ok...
>> ➜ kafka_2.10-0.10.0.0 bin/kafka-console-consumer.sh --new-consumer
>> --bootstrap-server localhost:9092 --topic streams-wordcount-Counts-changelog
>> --property print.key=true --property key.separator=":" --from-beginning
>> *msg2:*
>> *msg4:*
>> *msg6:*
>> *msg1:*
>> *msg3:*
>> *msg5:*
>> Everything is ok, except changelog-topic trouble me. Why it's value is
>> empty?
>>
>> I have dig into source code, and summary the workflow :
>> 1. when WordCountProcessorDemo's ProcessorNode put kv to KeyValueStore,
>> no matter Memory or RocksDB, they both put into local storage, then append
>> msg to StoreChangeLogger
>> 2. the key append to StoreChangeLogger first, and invoke maybeLogChange
>> by passing ValueGetter, this getter will get value from local storage when
>> logChange() operation happen
>> 3. when logChange() on StoreChangeLogger happen, send KeyValue message to
>> changelog topic, here is streams-wordcount-Counts-changelog
>> 4. StoreChangeLogger use dirty and remove Set to swap between logChange()
>> and add().
>>
>> for (K k : this.dirty) { // logChange() method flush dirty to changelog topic
>> V v = getter.get(k); // value getter will get value from local storage
>> collector.send(new ProducerRecord<>(this.topic, this.partition, k, v),
>> keySerializer, valueSerializer);
>> }
>>
>>
>> The Only reason I can figure out why value is empty is when invoke
>> KeyValueStore's delete method. But Actualy the application did't do it
>>
>> Anyone help me, Tks .
>>
>>
>
