ATLAS-3796: Status Reporter display fix.

intg/src/main/java/org/apache/atlas/pc/StatusReporter.java

@@ -98,7 +98,7 @@ public class StatusReporter<T, U> {
     private U acknowledged(Map.Entry<T, U> lookFor) {
         U ack = lookFor.getValue();
         producedItems.remove(lookFor.getKey());
-        processedSet.remove(lookFor);
+        processedSet.remove(lookFor.getKey());
         return ack;
     }
 
@@ -107,4 +107,12 @@ public class StatusReporter<T, U> {
         lastAck = System.currentTimeMillis();
         return b;
     }
+
+    public int getProducedCount() {
+        return this.producedItems.size();
+    }
+
+    public int getProcessedCount() {
+        return this.processedSet.size();
+    }
 }

intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java

@@ -82,6 +82,9 @@ public class StatusReporterTest {
         extractResults(wi, statusReporter);
         assertEquals(statusReporter.ack().intValue(), (maxItems - 1));
         wi.shutdown();
+
+        assertEquals(statusReporter.getProducedCount(), 0);
+        assertEquals(statusReporter.getProcessedCount(), 0);
     }
 
     private void extractResults(WorkItemManager<Integer, WorkItemConsumer> wi, StatusReporter<Integer, Integer> statusReporter) {