Balance Alerts Topology

 KStream<CustomerId, KeyValue<SpecificRecord, OutboundMessage>> addressedMessages =
                builder.<AccountId, AccountEntry>stream(accountEntryStream)
                        .leftJoin(accountToCustomerIds, (accountEntry, customerIds) -> {
                            if (isNull(customerIds)) {
                                return Collections.<KeyValue<CustomerId, AccountEntry>>emptyList();
                            } else {
                                return customerIds.getCustomerIds().stream()
                                        .map(customerId -> KeyValue.pair(customerId, accountEntry))
                                        .collect(toList());
                            }
                        })
                        .flatMap((accountId, accountentryByCustomer) -> accountentryByCustomer)
                        .through(customerIdToAccountEntryStream)
                        .leftJoin(alertSettings, Pair::with)
                        .flatMapValues(
                                (Pair<AccountEntry, CustomerAlertSettings> accountEntryAndSettings) ->
                                        BalanceAlertsGenerator.generateAlerts(
                                                accountEntryAndSettings.getValue0(),
                                                accountEntryAndSettings.getValue1())
                        );