summaryrefslogtreecommitdiffstats
path: root/src/main
diff options
context:
space:
mode:
Diffstat (limited to 'src/main')
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java4
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java4
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPCambriaLimiter.java2
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java2
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java2
5 files changed, 7 insertions, 7 deletions
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
index 4e54a3c..67a8af9 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
@@ -113,13 +113,13 @@ public class Kafka011Consumer implements Consumer {
public synchronized Consumer.Message nextMessage() {
try {
- if (fPendingMsgs.isEmpty()) {
+ if (!fPendingMsgs.isEmpty()) {
return makeMessage(fPendingMsgs.take());
}
} catch (InterruptedException x) {
log.warn("After size>0, pending msg take() threw InterruptedException. Ignoring. (" + x.getMessage() + ")",
x);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
Callable<Boolean> run = new Callable<Boolean>() {
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
index 00b6a75..647f4ce 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
@@ -299,7 +299,7 @@ public class KafkaConsumerCache {
curator.blockUntilConnected();
} catch (InterruptedException e) {
log.error("error while setting curator framework :",e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
}
@@ -512,7 +512,7 @@ public class KafkaConsumerCache {
Thread.sleep(consumerHandoverWaitMs);
} catch (InterruptedException e) {
log.error("InterruptedException in dropTimedOutConsumer",e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
log.info("Dropped " + key + " consumer due to timeout");
}
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPCambriaLimiter.java b/src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPCambriaLimiter.java
index 01c46e7..f645c8d 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPCambriaLimiter.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPCambriaLimiter.java
@@ -154,7 +154,7 @@ public class DMaaPCambriaLimiter {
}
} catch (InterruptedException e) {
log.error("Exception "+ e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java b/src/main/java/org/onap/dmaap/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java
index 9c1fcb2..f59f704 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java
@@ -198,7 +198,7 @@ public class DMaaPCambriaSimplerBatchPublisher extends CambriaBaseClient
}
} catch (InterruptedException e) {
getLog().warn("Possible message loss. " + e.getMessage(), e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
} catch (IOException e) {
getLog().warn("Possible message loss. " + e.getMessage(), e);
}
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java b/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
index 27ffc89..537fc22 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
@@ -333,7 +333,7 @@ public class CambriaOutboundEventStream implements StreamWriter {
// FIXME: would be good to wait/signal
Thread.sleep(100);
} catch (InterruptedException e) {
- // Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
}
});