Merge pull request #2479 from telefonicaid/fix/channel_usage

Fix/channel usage

Author: fgalan

CHANGES_NEXT_RELEASE

@@ -1 +1 @@
-
+- [cygnus-ngsi] Fix log about channel usage when no CygnusChannel type is used (bug introduced in 3.17.0) (#2478)

cygnus-common/src/main/java/com/telefonica/iot/cygnus/channels/CygnusFileChannel.java

@@ -120,8 +120,11 @@ public void setNumTakesFail(long n) {
     public double getUsage() {
         long currentSize = channelCounterRef.getChannelSize();
         long maxCapacity = channelCounterRef.getChannelCapacity();
-        double usagePercentage = (double) currentSize / maxCapacity * 100;
-        return usagePercentage;
+        if (currentSize < 0 || maxCapacity <= 0) {
+            return 0.0;
+        }
+
+        return (currentSize * 100.0) / maxCapacity;
     } // getUsage
 
 } // CygnusFileChannel

cygnus-common/src/main/java/com/telefonica/iot/cygnus/channels/CygnusMemoryChannel.java

@@ -124,8 +124,11 @@ public void setNumTakesFail(long n) {
     public double getUsage() {
         long currentSize = channelCounterRef.getChannelSize();
         long maxCapacity = channelCounterRef.getChannelCapacity();
-        double usagePercentage = (double) currentSize / maxCapacity * 100;
-        return usagePercentage;
+        if (currentSize < 0 || maxCapacity <= 0) {
+            return 0.0;
+        }
+
+        return (currentSize * 100.0) / maxCapacity;
     } // getUsage
 
 } // CygnusMemoryChannel

cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSISink.java

@@ -514,11 +514,17 @@ private Status processNewBatches() {
         try {
             txn.begin();
 
-            CygnusChannel cygnusch = (CygnusChannel) ch;
-            channelUsage = cygnusch.getUsage();
-            LOGGER.debug("Channel usage: " + channelUsage + "% in sink " + this.getName());
-            if (channelUsage > NGSIConstants.HIGH_CHANNEL_PERCENT_USAGE) {
-                LOGGER.warn("High Channel usage: " + channelUsage + "% in sink " + this.getName());
+            if (ch instanceof CygnusChannel) {
+                CygnusChannel cygnusch = (CygnusChannel) ch;
+                channelUsage = cygnusch.getUsage();
+
+                LOGGER.debug("Channel usage: " + channelUsage + "% in sink " + this.getName());
+                if (channelUsage > NGSIConstants.HIGH_CHANNEL_PERCENT_USAGE) {
+                    LOGGER.warn("High Channel usage: " + channelUsage + "% in sink " + this.getName());
+                }
+            } else {
+                // Flume channel "vanilla": do not get usage
+                LOGGER.debug("Channel is not a CygnusChannel (" + ch.getClass().getName() + "). Skipping channel usage.");
             }
 
             // Get and process as many events as the batch size