我想知道，在我与kafka合作的架构中，当新的东西到达我的代理时，是否有可能执行这些流。例如，下面的体系结构。我不想安排这些任务，是的，每次新的流到达时，我都要执行此处理。