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