KStreams-如何处理一个主题的消息延迟

Question

我有一个基于Spring boot的KStreams应用程序，我在其中加入跨多个主题的数据。当一个主题出现延迟时，处理情况的最佳实践是什么？我读过一些链接，比如如何管理Kafka KStream到KStream窗口连接？和其他人。

下面是我的示例代码（Spring Boot应用程序），用于为两个主题--雇员和财务--生成模拟数据。下面是员工主题的代码：

private void sendEmpData() {
    IntStream.range(0, 1).forEach(index -> {
        EmployeeKey key = new EmployeeKey();
        key.setEmployeeId(1);

        Employee employee = new Employee();
        employee.setDepartmentId(1000);
        employee.setEmployeeFirstName("John);
        employee.setEmployeeId(1);
        employee.setEmployeeLastName("Doe");

        kafkaTemplateForEmp.send(EMP_TOPIC, key, employee);
    });
}

对于金融主题也是如此：

private void sendFinanceData() {
    IntStream.range(0, 1).forEach(index -> {
        FinanceKey key = new FinanceKey();
        key.setEmployeeId(1);
        key.setDepartmentId(1000);

        Finance finance = new Finance();
        finance.setDepartmentId(1000);
        finance.setEmployeeId(1);
        finance.setSalary(2000);

        kafkaTemplateForFinance.send(FINANCE_TOPIC, key, finance);
    });
}

employeeKStream.join(reKeyedStream,
            (employee, finance) -> new EmployeeFinance(employee.getEmployeeId(),
                    employee.getEmployeeFirstName(),
                    employee.getEmployeeLastName(),
                    employee.getDepartmentId(),
                    finance.getSalary(),
                    finance.getSalaryGrade()),
            JoinWindows.of(windowRetentionTimeMs), //30 seconds
            Joined.with(
                    employeeKeySerde,
                    employeeSerde,
                    financeSerde)).to(outputTopic, Produced.with(employeeKeySerde, employeeFinanceSerde));

Answer 1

如果在Financial主题中有一个匹配键的记录在30秒后到达，那么连接就不会发生。

默认情况下，Kafka Streams使用24小时的宽限期，因此，即使存在滞后或无序的数据，您的联接也应该工作。注意，Kafka中的滞后总是指读取路径！

在财经话题30秒后到达