Я новичок в Kafka / Kafka Stream. Я использую последний Кафка / Кафка-поток и Кафка-клиент и openjdk11 . Мой продюсер производит объекты json (где key - это name ), которое выглядит как
{"Name":"John", "amount":123, "time":2019-10-03T05:24:52" }
Код производителя для лучшего понимания:
public static ProducerRecord<String, String> newRandomTransaction(String name) {
// creates an empty json {}
ObjectNode transaction = JsonNodeFactory.instance.objectNode();
Integer amount = ThreadLocalRandom.current().nextInt(0, 100);
// Instant.now() is to get the current time
Instant now = Instant.now();
// we write the data to the json document
transaction.put("name", name);
transaction.put("amount", amount);
transaction.put("time", now.toString());
return new ProducerRecord<>("bank-transactions", name, transaction.toString());
}
Теперь я пытаюсь написать свое приложение, которое потребляет транзакции, и вычислить общую сумму на балансе этого человека.
( К вашему сведению: Я использую старуюкод и пытается заставить его работать).
Используется GroupBYKey , так как тема уже имеет правильный ключ. А затем агрегат для вычисления общего баланса , где я борюсь.
Приложение в данный момент (закомментированная часть - это старый код, который я пытаюсь создатьработа в следующей строке):
public class BankBalanceExactlyOnceApp {
private static ObjectMapper mapper = new ObjectMapper();
public static void main(String[] args) {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "bank-balance-application");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// we disable the cache to demonstrate all the "steps" involved in the transformation - not recommended in prod
config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0");
// Exactly once processing!!
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
// json Serde
final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);
StreamsBuilder builder = new StreamsBuilder();
KStream<String, JsonNode> bankTransactions =
builder.stream( "bank-transactions", Materialized.with(Serdes.String(), jsonSerde);
// create the initial json object for balances
ObjectNode initialBalance = JsonNodeFactory.instance.objectNode();
initialBalance.put("count", 0);
initialBalance.put("balance", 0);
initialBalance.put("time", Instant.ofEpochMilli(0L).toString());
/*KTable<String, JsonNode> bankBalance = bankTransactions
.groupByKey(Serdes.String(), jsonSerde)
.aggregate(
() -> initialBalance,
(key, transaction, balance) -> newBalance(transaction, balance),
jsonSerde,
"bank-balance-agg"
);*/
KTable<String, JsonNode> bankBalance = bankTransactions
.groupByKey(Serialized.with(Serdes.String(), jsonSerde))
.aggregate(
() -> initialBalance,
(key, transaction, balance) -> {
//String t = transaction.toString();
newBalance(transaction, balance);
},
Materialized.with(Serdes.String(), jsonSerde),
"bank-balance-agg"
);
bankBalance.toStream().to("bank-balance-exactly-once", Produced.with(Serdes.String(), jsonSerde));
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.cleanUp();
streams.start();
// print the topology
System.out.println(streams.toString());
// shutdown hook to correctly close the streams application
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
private static JsonNode newBalance(JsonNode transaction, JsonNode balance) {
// create a new balance json object
ObjectNode newBalance = JsonNodeFactory.instance.objectNode();
newBalance.put("count", balance.get("count").asInt() + 1);
newBalance.put("balance", balance.get("balance").asInt() + transaction.get("amount").asInt());
Long balanceEpoch = Instant.parse(balance.get("time").asText()).toEpochMilli();
Long transactionEpoch = Instant.parse(transaction.get("time").asText()).toEpochMilli();
Instant newBalanceInstant = Instant.ofEpochMilli(Math.max(balanceEpoch, transactionEpoch));
newBalance.put("time", newBalanceInstant.toString());
return newBalance;
}
}
Проблема: как я пытаюсь вызвать newBalance (транзакция, баланс) в строке:
aggregate(
() -> initialBalance,
(key, transaction, balance) -> newBalance(transaction, balance),
jsonSerde,
"bank-balance-agg"
)
и увидев компилятор ошибка с сообщением:
newBalance(JsonNode, JsonNode) can not be applied to (<lambda parameter>,<lambda parameter>)
Я попытался прочитать его как строку, изменил тип параметра с JsonNode на Object. Тем не менее, не могу это исправить.
Могу ли я получить какие-либо предложения о том, как это исправить?