Kafka流聚合删除和/或密钥更改

Question

        ...
        KStream<String, Employee> stream = kStreamBuilder.stream("EMPLOYEE"); // Stream from raw EMPLOYEE
        stream.map((k, v) -> new KeyValue<>(k, transformEmployee(v))) // <-- some stateless enrichment of the employee
                .groupBy((k, emp) -> emp.getDepartmentId(), jsonSerialisedWith(Employee.class))

                // dummy reduce to a get a ktable for agg:
                .reduce((aggValue, newEmp) -> newEmp) 
                .groupBy((k, emp2) -> new KeyValue<>(emp2.getDepartmentId(), emp2), jsonSerialisedWith(Employee.class))

                .aggregate(Department::new, this::addEmployee, this::removeEmployee,
                           jsonValueMaterializedAs("DEPARTMENT-AGG", Department.class))
                .toStream()
                .to("DEPARTMENT", jsonProducedWith(Department.class));
        ...

    private Department addEmployee(String deptId, Employee employee, Department department) {
        department.addEmployee(employee);
        if (department.getId() == null) {
            department.setId(employee.getDepartmentId());
            department.setName(employee.getDepartmentName());
        }
        return department;
    }

这适用于添加或更新。但是，随着时间的推移，员工可能会被删除或重新分配到另一个部门。我认为删除应该是发送到EMPLOYEE主题的tombstone记录（k：empid,v：null）。但是，我不再拥有departmentId，我必须进行空检查（并为departmentId返回空），因此删除员工时不会发生removeEmployee。DepartmentID的更改也有类似问题。

那么，Kafka的方法是什么呢？

Answer 1

我认为使用您的代码就足够了，但是稍微改变一下删除该员工的语义。

您应该添加某种mock部门（将在用户从部门中删除时使用）。

如果删除了employee,则将department设置为null,则应将其分配给mockdepartment。