-
- for (KafkaJobDataConsumer consumer : consumers.values()) {
- if (!consumer.isRunning()) {
- restartTopic(consumer);
- }
- }
+ this.consumers.keySet().forEach(typeId -> {
+ this.consumers.get(typeId).forEach(consumer -> {
+ if (!consumer.isRunning()) {
+ restartTopic(consumer);
+ }
+ });
+ });