Snippet of TransfersRecordingServiceJoinStreams.java

TransfersRecordingServiceJoinStreams.java

        final KStream<String, Transfer> transfersKStream = 
                createInputKStream(stringSerde, configValues.getSchemaRegistryUrl(), 
                builder);

        final Serde<AccountBalance> accountBalanceSerde = 
                createAccountBalanceSerde(configValues.getSchemaRegistryUrl());        
        
        final KTable<String,AccountBalance> accountBalancesKTable = 
                createInputKTable(stringSerde, accountBalanceSerde, builder);
        
        final KStream<String, AccountBalance> newABstream = transfersKStream.join(
                accountBalancesKTable, new TransferAndAccountBalanceJoiner());
        
        newABstream.to(ACCOUNT_BALANCES_TOPIC, Produced.with(stringSerde, accountBalanceSerde));