Output msgId only when logging ack/fails.

Updated issue apache#2909 to only output the message ID for log messages
produced by nextTuple, ack, and fail within SupplierSource and
ComplexSource.

Author: jmark99

heron/api/src/java/org/apache/heron/streamlet/impl/sources/ComplexSource.java

@@ -82,7 +82,7 @@ public void nextTuple() {
           collector.emit(new Values(tuple));
         }
         // TODO change logging level
-        LOG.info("Emitting: " + new Values(tuple, msgId));
+        LOG.info("Emitting: [" + msgId + "]");
       }
     }
   }
@@ -100,7 +100,7 @@ public void nextTuple() {
       Values values = new Values(msgIdCache.getIfPresent(mid));
       collector.emit(values, mid);
       // TODO change logging level
-      LOG.info("Re-emit:  [" + values.get(0) + ", " + mid + "]");
+      LOG.info("Re-emit:  [" + mid + "]");
     }
   }
 }

heron/api/src/java/org/apache/heron/streamlet/impl/sources/SupplierSource.java

@@ -66,7 +66,7 @@ public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector
       collector.emit(new Values(data));
     }
     // TODO change logging level
-    LOG.info("Emitting: " + new Values(data, msgId));
+    LOG.info("Emitting: [" + msgId + "]");
   }
 
   @Override public void ack(Object mid) {
@@ -82,7 +82,7 @@ public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector
       Values values = new Values(msgIdCache.getIfPresent(mid));
       collector.emit(values, mid);
       // TODO change logging level
-      LOG.info("Re-emit:  [" + values.get(0) + ", " + mid + "]");
+      LOG.info("Re-emit:  [" + mid + "]");
     }
   }
 }