KStream Присоединяйтесь к GlobalKTable по неключевым значениям

Question

Я пытаюсь присоединиться к KStream с помощью GlobalKTable, соединение не полностью выполняется по ключам.

GlobalKTable<String, Employee> employeesDetails = builder.globalTable("EMPLOYEE_TOPIC",..);
KStream<String,String> empIdOverLoginUserId = builder.stream("LOG_TOPIC", ….);

Я хочу объединить empIdOverLoginUserId со служащим по отношению к значению через empIdOverLoginUserId по ключу employeeDetails.

Любое ключ?

Answer 1

Второй параметр соединения KStream-GlobalKTable - это KeyValueMapper для сопоставления записи KStream (key-value) с ключом GlobalKTable, к которому вы хотите присоединиться, вы можете использовать его для использования значения empIdOverLoginUserId в качестве ключа при объединении с GlobalKTable:

empIdOverLoginUserId.join(
                employeesDetails, 
                (userKey, userValue) -> userValue, 
                (userValue, employeesDetailValue) -> employeesDetailValue)