- public synchronized void restartNonRunningTasks() {
- this.consumers.keySet().forEach(typeId -> {
- this.consumers.get(typeId).forEach(consumer -> {
- if (!consumer.isRunning()) {
- restartTopic(consumer);
- }
- });
- });
+ public synchronized void restartNonRunningTopics() {
+ for (String typeId : this.consumers.keySet()) {
+ for (KafkaJobDataConsumer consumer : this.consumers.get(typeId)) {
+ restartTopic(consumer);
+ }
+ }