props.put("auto.offset.reset", properties.getAutoOffsetReset());
props.put("metadata.max.age.ms", "1000");
props.put("key.deserializer", StringDeserializer.class.getName());
- props.put("value.deserializer", JsonDeserializer.class.getName());
- props.put(JsonDeserializer.TRUSTED_PACKAGES, "de.juplo.kafka");
- props.put(JsonDeserializer.TYPE_MAPPINGS,
- Message.Type.ADD + ":" + MessageAddNumber.class.getName() + "," +
- Message.Type.CALC + ":" + MessageCalculateSum.class.getName());
+ props.put("value.deserializer", "TODO: JsonDeserializer konfigurieren");
return new KafkaConsumer<>(props);
}
String user = record.key();
Message message = record.value();
- switch(message.getType())
- {
- case ADD:
- addNumber(partition, user, (MessageAddNumber) message);
- break;
-
- case CALC:
- calculateSum(partition, user, (MessageCalculateSum) message);
- break;
- }
+ // TODO: JSON-Nachrichten verarbeiten
if (throttle.isPresent())
{