[streaming] Shorter log messages

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/FaultToleranceBuffer.java

@@ -196,7 +196,6 @@ public class FaultToleranceBuffer {
 	 */
 	public void failRecord(String recordID) {
 		// Create new id to avoid double counting acks
-		log.warn("Fail ID: " + recordID);
 		StreamRecord newRecord = removeRecord(recordID).setId(channelID);
 		addRecord(newRecord);
 		reEmit(newRecord);

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/JobGraphBuilder.java

@@ -90,7 +90,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", sourceName);
 		components.put(sourceName, source);
 		numberOfInstances.put(sourceName, 1);
-		log.debug("Source set: " + sourceName);
+		log.debug("SOURCE: " + sourceName);
 	}
 
 	/**
@@ -116,7 +116,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", sourceName);
 		components.put(sourceName, source);
 		numberOfInstances.put(sourceName, 1);
-		log.debug("Source set: " + sourceName);
+		log.debug("SOURCE: " + sourceName);
 	}
 
 	/**
@@ -138,7 +138,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", taskName);
 		components.put(taskName, task);
 		numberOfInstances.put(taskName, 1);
-		log.debug("Task set: " + taskName);
+		log.debug("TASK: " + taskName);
 	}
 
 	/**
@@ -164,7 +164,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", taskName);
 		components.put(taskName, task);
 		numberOfInstances.put(taskName, parallelism);
-		log.debug("Task set: " + taskName);
+		log.debug("TASK: " + taskName);
 	}
 
 	/**
@@ -186,7 +186,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", sinkName);
 		components.put(sinkName, sink);
 		numberOfInstances.put(sinkName, 1);
-		log.debug("Sink set: " + sinkName);
+		log.debug("SINK: " + sinkName);
 	}
 
 	/**
@@ -212,7 +212,7 @@ public class JobGraphBuilder {
 		config.setString("componentName", sinkName);
 		components.put(sinkName, sink);
 		numberOfInstances.put(sinkName, 1);
-		log.debug("Sink set: " + sinkName);
+		log.debug("TASK: " + sinkName);
 	}
 
 	/**
@@ -247,14 +247,14 @@ public class JobGraphBuilder {
 					"partitionerClass_"
 							+ upStreamComponent.getNumberOfForwardConnections(),
 					PartitionerClass);
-			log.debug("Components connected with "
-					+ PartitionerClass.getSimpleName() + ": "
-					+ upStreamComponentName + " to " + downStreamComponentName);
+			log.debug("CONNECTED: "
+					+ PartitionerClass.getSimpleName() + " - "
+					+ upStreamComponentName + " -> " + downStreamComponentName);
 		} catch (JobGraphDefinitionException e) {
 			log.error(
 					"Cannot connect components with "
 							+ PartitionerClass.getSimpleName() + " : "
-							+ upStreamComponentName + " to "
+							+ upStreamComponentName + " -> "
 							+ downStreamComponentName, e);
 		}
 	}
@@ -338,8 +338,8 @@ public class JobGraphBuilder {
 					keyPosition);
 
 			addOutputChannels(upStreamComponentName);
-			log.debug("Components connected by field: " + upStreamComponentName
-					+ " to " + downStreamComponentName + " by key position "
+			log.debug("CONNECTED: FIELD PARTITIONING - " + upStreamComponentName
+					+ " -> " + downStreamComponentName + ", KEY: "
 					+ keyPosition);
 		} catch (JobGraphDefinitionException e) {
 			log.error("Cannot connect components by field: "

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamComponentHelper.java

@@ -180,7 +180,7 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 			} else {
 				partitioners.add(partitioner.newInstance());
 			}
-			log.debug("Partitioner set: " + partitioner.getSimpleName() + " with "
+			log.trace("Partitioner set: " + partitioner.getSimpleName() + " with "
 					+ nrOutput + " outputs");
 		} catch (Exception e) {
 			log.error(

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamSink.java

@@ -60,7 +60,7 @@ public class StreamSink extends AbstractOutputTask {
 
 	@Override
 	public void invoke() throws Exception {
-		log.debug("Sink " + name + " invoked");
+		log.debug("SINK " + name + " invoked");
 		boolean hasInput = true;
 		while (hasInput) {
 			hasInput = false;
@@ -82,6 +82,6 @@ public class StreamSink extends AbstractOutputTask {
 			}
 		}
 		System.out.println("Result: "+userFunction.getResult());
-		log.debug("Sink " + name + " invoke finished");
+		log.debug("SINK " + name + " invoke finished");
 	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamSource.java

@@ -84,7 +84,7 @@ public class StreamSource extends AbstractInputTask<RandIS> {
 
 	@Override
 	public void invoke() throws Exception {
-		log.debug("Source " + name + " invoked with instance id " + sourceInstanceID);
+		log.debug("SOURCE " + name + " invoked with instance id " + sourceInstanceID);
 		userFunction.invoke();
 	}
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamTask.java

@@ -81,7 +81,7 @@ public class StreamTask extends AbstractTask {
 
 	@Override
 	public void invoke() throws Exception {
-		log.debug("Task " + name + " invoked with instance id " + taskInstanceID);
+		log.debug("TASK " + name + " invoked with instance id " + taskInstanceID);
 
 		boolean hasInput = true;
 		while (hasInput) {
@@ -98,12 +98,12 @@ public class StreamTask extends AbstractTask {
 						log.debug("ACK: " + id + " -- " + name);
 					} catch (Exception e) {
 						streamTaskHelper.threadSafePublish(new FailEvent(id), input);
-						log.warn("INVOKE FAILED: " + id + " -- " + name + " -- due to " + e.getMessage());
+						log.warn("FAILED: " + id + " -- " + name + " -- due to " + e.getMessage());
 					}
 				}
 			}
 		}
-		log.debug("Task " + name + "invoke finished with instance id " + taskInstanceID);
+		log.debug("TASK " + name + "invoke finished with instance id " + taskInstanceID);
 	}
 
 }