+ log.debug("polled {} records", records.count());
+ records.forEach(record -> handleRecord(record, productionUseCases));
+ }
+ catch (WakeupException e)
+ {
+ log.info("cleanly interrupted while polling");
+ }
+ }
+
+ log.info("polling stopped");
+ }
+
+ private void handleRecord(ConsumerRecord<String, String> record, ConsumerUseCases useCases)
+ {
+ try
+ {
+ byte eventType = record.headers().lastHeader(EventType.HEADER).value()[0];
+
+ switch (eventType)
+ {
+ case EventType.NEW_TRANSFER:
+
+ NewTransferEvent newTransferEvent =
+ mapper.readValue(record.value(), NewTransferEvent.class);
+ useCases
+ .create(
+ newTransferEvent.getId(),
+ newTransferEvent.getPayer(),
+ newTransferEvent.getPayee(),
+ newTransferEvent.getAmount());
+ break;
+
+ case EventType.TRANSFER_STATE_CHANGED:
+
+ TransferStateChangedEvent stateChangedEvent =
+ mapper.readValue(record.value(), TransferStateChangedEvent.class);
+ useCases.handleStateChange(stateChangedEvent.getId(), stateChangedEvent.getState());
+ break;
+ }
+ }
+ catch (JsonProcessingException e)
+ {
+ log.error(
+ "ignoring invalid json in message #{} on {}/{}: {}",
+ record.offset(),
+ record.topic(),
+ record.partition(),
+ record.value());
+ }
+ catch (IllegalArgumentException e)
+ {
+ log.error(
+ "ignoring invalid message #{} on {}/{}: {}, message={}",
+ record.offset(),
+ record.topic(),
+ record.partition(),
+ e.getMessage(),
+ record.value());
+ }
+ }
+
+
+ public Optional<String> uriForKey(String key)
+ {
+ synchronized (this)
+ {
+ while (partitionOwnershipUnknown)
+ {
+ try { wait(); } catch (InterruptedException e) {}
+ }
+
+ int partition = TransferPartitioner.computeHashForKey(key, numPartitions);
+ return
+ Optional
+ .ofNullable(instanceIdByPartition[partition])
+ .map(id -> instanceIdUriMapping.get(id));
+ }
+ }
+
+ @EventListener
+ public synchronized void onApplicationEvent(ContextRefreshedEvent event)
+ {
+ // "Needed", because this method is called synchronously during the
+ // initialization pahse of Spring. If the subscription happens
+ // in the same thread, it would block the completion of the initialization.
+ // Hence, the app would not react to any signal (CTRL-C, for example) except
+ // a KILL until the restoring is finished.
+ future = CompletableFuture.runAsync(() -> start());
+ log.info("start of application completed");
+ }
+
+
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition> partitions)
+ {
+ partitionOwnershipUnknown = true;
+ log.info("partitions revoked: {}", partitions);
+ }
+
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition> partitions)
+ {
+ log.info("partitions assigned: {}", partitions);
+ fetchAssignmentsAsync();
+ if (partitions.size() > 0)
+ restore(partitions);
+ }
+
+ private void fetchAssignmentsAsync()
+ {
+ adminClient
+ .describeConsumerGroups(List.of(groupId))
+ .describedGroups()
+ .get(groupId)
+ .whenComplete((descriptions, e) ->