X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRecordHandler.java;h=51d524fc682beb6a8998075a55bf037c7c25ee51;hb=627763878b235ba168c7f55a1ef448851b027bfc;hp=ce340a76a96eaec73853fb42d67fd60e7a62e665;hpb=d576eea9bc9208d9e5003bd8c8c132bed96b5c40;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java index ce340a7..51d524f 100644 --- a/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java +++ b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java @@ -1,15 +1,23 @@ package de.juplo.kafka; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; +import java.time.Duration; import java.util.HashMap; import java.util.Map; +import java.util.Optional; +@RequiredArgsConstructor @Slf4j public class ApplicationRecordHandler implements RecordHandler { + private final AdderResults results; + private final Optional throttle; + private final String id; + private final Map state = new HashMap<>(); @@ -22,20 +30,34 @@ public class ApplicationRecordHandler implements RecordHandler if (message.equals("CALCULATE")) { - Long result = state.get(partition).calculate(user); - log.info("New result for {}: {}", user, result); - return; + AdderResult result = state.get(partition).calculate(user); + log.info("{} - New result for {}: {}", id, user, result); + results.addResults(partition, user, result); + } + else + { + state.get(partition).addToSum(user, Integer.parseInt(message)); } - state.get(partition).addToSum(user, Integer.parseInt(message)); + if (throttle.isPresent()) + { + try + { + Thread.sleep(throttle.get().toMillis()); + } + catch (InterruptedException e) + { + log.warn("{} - Intrerrupted while throttling: {}", id, e); + } + } } - protected void addPartition(Integer partition, Map state) + protected void addPartition(Integer partition, Map state) { this.state.put(partition, new AdderBusinessLogic(state)); } - protected Map removePartition(Integer partition) + protected Map removePartition(Integer partition) { return this.state.remove(partition).getState(); } @@ -45,4 +67,9 @@ public class ApplicationRecordHandler implements RecordHandler { return state; } + + public AdderBusinessLogic getState(Integer partition) + { + return state.get(partition); + } }