Skip to content
Snippets Groups Projects
Commit 48e1879c authored by Björn Vonheiden's avatar Björn Vonheiden
Browse files

Fix use of key serializer instead value serializer of avro

parent b78c1326
No related branches found
No related tags found
2 merge requests!28Use Titan CC Avro Records in UC App and Workload Generator,!13Migrate to new Titan CC records
......@@ -40,7 +40,7 @@ public class TopologyBuilder {
this.builder
.stream(this.inputTopic, Consumed.with(
Serdes.String(),
this.srAvroSerdeFactory.<ActivePowerRecord>forKeys()))
this.srAvroSerdeFactory.<ActivePowerRecord>forValues()))
.mapValues(v -> this.gson.toJson(v))
.foreach((k, v) -> LOGGER.info("Key: " + k + " Value: " + v));
......
......@@ -15,7 +15,7 @@ import titan.ccp.common.kafka.avro.SchemaRegistryAvroSerdeFactory;
/**
* Sends monitoring records to Kafka.
*
* @param <T> {@link IMonitoringRecord} to send
* @param <T> {@link SpecificRecord} to send
*/
public class KafkaRecordSender<T extends SpecificRecord> {
......@@ -72,7 +72,7 @@ public class KafkaRecordSender<T extends SpecificRecord> {
this.producer = new KafkaProducer<>(properties,
new StringSerializer(),
srAvroSerdeFactory.<T>forKeys().serializer());
srAvroSerdeFactory.<T>forValues().serializer());
}
/**
......
......@@ -50,7 +50,7 @@ public class TopologyBuilder {
this.builder
.stream(this.inputTopic,
Consumed.with(Serdes.String(),
this.srAvroSerdeFactory.<ActivePowerRecord>forKeys()))
this.srAvroSerdeFactory.<ActivePowerRecord>forValues()))
.groupByKey()
.windowedBy(TimeWindows.of(this.duration))
// .aggregate(
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment