juplo.de Git - demos/kafka/demos-kafka-payment-system-transfer/blob - src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferConsumer.java

   1 package de.juplo.kafka.payment.transfer.adapter;
   2
   3 import com.fasterxml.jackson.core.JsonProcessingException;
   4 import com.fasterxml.jackson.databind.ObjectMapper;
   5 import de.juplo.kafka.payment.transfer.ports.CreateTransferUseCase;
   6 import de.juplo.kafka.payment.transfer.ports.GetTransferUseCase;
   7 import de.juplo.kafka.payment.transfer.ports.HandleStateChangeUseCase;
   8 import de.juplo.kafka.payment.transfer.ports.TransferRepository;
   9 import lombok.extern.slf4j.Slf4j;
  10 import org.apache.kafka.clients.admin.AdminClient;
  11 import org.apache.kafka.clients.admin.MemberDescription;
  12 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
  13 import org.apache.kafka.clients.consumer.ConsumerRecord;
  14 import org.apache.kafka.clients.consumer.ConsumerRecords;
  15 import org.apache.kafka.clients.consumer.KafkaConsumer;
  16 import org.apache.kafka.common.TopicPartition;
  17 import org.apache.kafka.common.errors.WakeupException;
  18 import org.springframework.context.event.ContextRefreshedEvent;
  19 import org.springframework.context.event.EventListener;
  20 import org.springframework.web.bind.annotation.PostMapping;
  21 import org.springframework.web.bind.annotation.RequestMapping;
  22 import org.springframework.web.bind.annotation.ResponseBody;
  23
  24 import java.time.Duration;
  25 import java.util.*;
  26 import java.util.concurrent.CompletableFuture;
  27 import java.util.concurrent.ExecutionException;
  28 import java.util.concurrent.Future;
  29 import java.util.stream.Collectors;
  30
  31
  32 @RequestMapping("/consumer")
  33 @ResponseBody
  34 @Slf4j
  35 public class TransferConsumer implements Runnable, ConsumerRebalanceListener
  36 {
  37   private final String topic;
  38   private final int numPartitions;
  39   private final KafkaConsumer<String, String> consumer;
  40   private final AdminClient adminClient;
  41   private final TransferRepository repository;
  42   private final ObjectMapper mapper;
  43   private final ConsumerUseCases productionUseCases, restoreUseCases;
  44
  45   private boolean running = false;
  46   private boolean shutdown = false;
  47   private Future<?> future = null;
  48
  49   private final String groupId;
  50   private final String groupInstanceId;
  51   private final Map<String, String> instanceIdUriMapping;
  52   private final String[] instanceIdByPartition;
  53
  54   private volatile boolean partitionOwnershipUnknown = true;
  55
  56
  57   public TransferConsumer(
  58       String topic,
  59       int numPartitions,
  60       Map<String, String> instanceIdUriMapping,
  61       KafkaConsumer<String, String> consumer,
  62       AdminClient adminClient,
  63       TransferRepository repository,
  64       ObjectMapper mapper,
  65       ConsumerUseCases productionUseCases,
  66       ConsumerUseCases restoreUseCases)
  67   {
  68     this.topic = topic;
  69     this.numPartitions = numPartitions;
  70     this.groupId = consumer.groupMetadata().groupId();
  71     this.groupInstanceId = consumer.groupMetadata().groupInstanceId().get();
  72     this.instanceIdByPartition = new String[numPartitions];
  73     this.instanceIdUriMapping = new HashMap<>(instanceIdUriMapping.size());
  74     for (String instanceId : instanceIdUriMapping.keySet())
  75     {
  76       // Requests are not redirected for the instance itself
  77       String uri = instanceId.equals(groupInstanceId)
  78           ? null
  79           : instanceIdUriMapping.get(instanceId);
  80       this.instanceIdUriMapping.put(instanceId, uri);
  81     }
  82     this.consumer = consumer;
  83     this.adminClient = adminClient;
  84     this.repository = repository;
  85     this.mapper = mapper;
  86     this.productionUseCases = productionUseCases;
  87     this.restoreUseCases = restoreUseCases;
  88   }
  89
  90
  91   @Override
  92   public void run()
  93   {
  94     while (running)
  95     {
  96       try
  97       {
  98         ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  99         if (records.count() == 0)
 100           continue;
 101
 102         log.debug("polled {} records", records.count());
 103         records.forEach(record -> handleRecord(record, productionUseCases));
 104       }
 105       catch (WakeupException e)
 106       {
 107         log.info("cleanly interrupted while polling");
 108       }
 109     }
 110
 111     log.info("polling stopped");
 112   }
 113
 114   private void handleRecord(ConsumerRecord<String, String> record, ConsumerUseCases useCases)
 115   {
 116     try
 117     {
 118       byte eventType = record.headers().lastHeader(EventType.HEADER).value()[0];
 119
 120       switch (eventType)
 121       {
 122         case EventType.NEW_TRANSFER:
 123
 124           NewTransferEvent newTransferEvent =
 125               mapper.readValue(record.value(), NewTransferEvent.class);
 126           useCases
 127               .create(
 128                   newTransferEvent.getId(),
 129                   newTransferEvent.getPayer(),
 130                   newTransferEvent.getPayee(),
 131                   newTransferEvent.getAmount());
 132           break;
 133
 134         case EventType.TRANSFER_STATE_CHANGED:
 135
 136           TransferStateChangedEvent stateChangedEvent =
 137               mapper.readValue(record.value(), TransferStateChangedEvent.class);
 138           useCases.handleStateChange(stateChangedEvent.getId(), stateChangedEvent.getState());
 139           break;
 140       }
 141     }
 142     catch (JsonProcessingException e)
 143     {
 144       log.error(
 145           "ignoring invalid json in message #{} on {}/{}: {}",
 146           record.offset(),
 147           record.topic(),
 148           record.partition(),
 149           record.value());
 150     }
 151     catch (IllegalArgumentException e)
 152     {
 153       log.error(
 154           "ignoring invalid message #{} on {}/{}: {}, message={}",
 155           record.offset(),
 156           record.topic(),
 157           record.partition(),
 158           e.getMessage(),
 159           record.value());
 160     }
 161   }
 162
 163
 164   public Optional<String> uriForKey(String key)
 165   {
 166     synchronized (this)
 167     {
 168       while (partitionOwnershipUnknown)
 169       {
 170         try { wait(); } catch (InterruptedException e) {}
 171       }
 172
 173       int partition = TransferPartitioner.computeHashForKey(key, numPartitions);
 174       return
 175           Optional
 176               .ofNullable(instanceIdByPartition[partition])
 177               .map(id -> instanceIdUriMapping.get(id));
 178     }
 179   }
 180
 181   @EventListener
 182   public synchronized void onApplicationEvent(ContextRefreshedEvent event)
 183   {
 184     // "Needed", because this method is called synchronously during the
 185     // initialization pahse of Spring. If the subscription happens
 186     // in the same thread, it would block the completion of the initialization.
 187     // Hence, the app would not react to any signal (CTRL-C, for example) except
 188     // a KILL until the restoring is finished.
 189     future = CompletableFuture.runAsync(() -> start());
 190   }
 191
 192
 193   @Override
 194   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
 195   {
 196     partitionOwnershipUnknown = true;
 197     log.info("partitions revoked: {}", partitions);
 198   }
 199
 200   @Override
 201   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
 202   {
 203     log.info("partitions assigned: {}", partitions);
 204     fetchAssignmentsAsync();
 205     if (partitions.size() > 0)
 206       restore(partitions);
 207   }
 208
 209   private void fetchAssignmentsAsync()
 210   {
 211     adminClient
 212         .describeConsumerGroups(List.of(groupId))
 213         .describedGroups()
 214         .get(groupId)
 215         .whenComplete((descriptions, e) ->
 216         {
 217           if (e != null)
 218           {
 219             log.error("could not fetch group data: {}", e.getMessage());
 220           }
 221           else
 222           {
 223             synchronized (this)
 224             {
 225               for (MemberDescription description : descriptions.members())
 226               {
 227                 description
 228                     .assignment()
 229                     .topicPartitions()
 230                     .forEach(tp -> instanceIdByPartition[tp.partition()] = description.groupInstanceId().get());
 231               }
 232               partitionOwnershipUnknown = false;
 233               notifyAll();
 234             }
 235           }
 236         });
 237   }
 238
 239   @Override
 240   public void onPartitionsLost(Collection<TopicPartition> partitions)
 241   {
 242     partitionOwnershipUnknown = true;
 243     log.info("partiotions lost: {}", partitions);
 244   }
 245
 246
 247   private void restore(Collection<TopicPartition> partitions)
 248   {
 249     log.info("--> starting restore...");
 250
 251     partitions
 252         .stream()
 253         .map(topicPartition -> topicPartition.partition())
 254         .forEach(partition -> repository.resetStorageForPartition(partition));
 255
 256     Map<Integer, Long> lastSeen =
 257         consumer
 258             .endOffsets(partitions)
 259             .entrySet()
 260             .stream()
 261             .collect(Collectors.toMap(
 262                 entry -> entry.getKey().partition(),
 263                 entry -> entry.getValue() - 1));
 264
 265     Map<Integer, Long> positions =
 266         lastSeen
 267             .keySet()
 268             .stream()
 269             .collect(Collectors.toMap(
 270                 partition -> partition,
 271                 partition -> 0l));
 272
 273     while (
 274         positions
 275             .entrySet()
 276             .stream()
 277             .map(entry -> entry.getValue() < lastSeen.get(entry.getKey()))
 278             .reduce(false, (a, b) -> a || b))
 279     {
 280       try
 281       {
 282         ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
 283         if (records.count() == 0)
 284           continue;
 285
 286         log.debug("polled {} records", records.count());
 287         records.forEach(record ->
 288         {
 289           handleRecord(record, restoreUseCases);
 290           positions.put(record.partition(), record.offset());
 291         });
 292       }
 293       catch(WakeupException e)
 294       {
 295         log.info("--> cleanly interrupted while restoring");
 296       }
 297     }
 298
 299     log.info("--> restore completed!");
 300   }
 301
 302   @PostMapping("start")
 303   public synchronized String start()
 304   {
 305     if (running)
 306     {
 307       log.info("already running!");
 308       return "Already running!";
 309     }
 310
 311     int foundNumPartitions = consumer.partitionsFor(topic).size();
 312     if (foundNumPartitions != numPartitions)
 313     {
 314       log.error(
 315           "unexpected number of partitions for topic {}: expected={}, found={}",
 316           topic,
 317           numPartitions,
 318           foundNumPartitions
 319           );
 320       return "Wrong number of partitions for topic " + topic + ": " + foundNumPartitions;
 321     }
 322
 323     consumer.subscribe(List.of(topic), this);
 324
 325     running = true;
 326     future = CompletableFuture.runAsync(this);
 327
 328     log.info("started");
 329     return "Started";
 330   }
 331
 332   @PostMapping("stop")
 333   public synchronized String stop()
 334   {
 335     if (!running)
 336     {
 337       log.info("not running!");
 338       return "Not running";
 339     }
 340
 341     running = false;
 342
 343     if (!future.isDone())
 344       consumer.wakeup();
 345
 346     log.info("waiting for the consumer...");
 347     try
 348     {
 349       future.get();
 350     }
 351     catch (InterruptedException|ExecutionException e)
 352     {
 353       log.error("Exception while joining polling task!", e);
 354       return e.getMessage();
 355     }
 356     finally
 357     {
 358       future = null;
 359       consumer.unsubscribe();
 360     }
 361
 362     log.info("stopped");
 363     return "Stopped";
 364   }
 365
 366   public synchronized void shutdown()
 367   {
 368     log.info("shutdown initiated!");
 369     shutdown = true;
 370     stop();
 371     log.info("closing consumer");
 372     consumer.close();
 373   }
 374
 375
 376
 377   public interface ConsumerUseCases
 378       extends
 379         GetTransferUseCase,
 380         CreateTransferUseCase,
 381         HandleStateChangeUseCase {};
 382 }