[FLINK-6828] Activate checkstyle for runtime/deployment

This closes #4066.

flink-runtime/pom.xml

@@ -432,7 +432,6 @@ under the License.
 						**/runtime/client/**,
 						**/runtime/clusterframework/**,       
 						**/runtime/concurrent/**,
-						**/runtime/deployment/**,
 						**/runtime/execution/**,
 						**/runtime/executiongraph/**,
 						**/runtime/heartbeat/**,

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java

@@ -31,9 +31,6 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
 import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
 import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.Serializable;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -41,7 +38,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
 /**
  * Deployment descriptor for a single input channel instance.
  *
- * <p> Each input channel consumes a single subpartition. The index of the subpartition to consume
+ * <p>Each input channel consumes a single subpartition. The index of the subpartition to consume
  * is part of the {@link InputGateDeploymentDescriptor} as it is the same for each input channel of
  * the respective input gate.
  *
@@ -51,7 +48,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
 public class InputChannelDeploymentDescriptor implements Serializable {
 
 	private static final long serialVersionUID = 373711381640454080L;
-	private static Logger LOG = LoggerFactory.getLogger(InputChannelDeploymentDescriptor.class);
 
 	/** The ID of the partition the input channel is going to consume. */
 	private final ResultPartitionID consumedPartitionId;
@@ -111,7 +107,7 @@ public class InputChannelDeploymentDescriptor implements Serializable {
 						producerState == ExecutionState.FINISHED ||
 						producerState == ExecutionState.SCHEDULED ||
 						producerState == ExecutionState.DEPLOYING)) {
-				
+
 				final TaskManagerLocation partitionTaskManagerLocation = producerSlot.getTaskManagerLocation();
 				final ResourceID partitionTaskManager = partitionTaskManagerLocation.getResourceID();
 

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java

@@ -32,7 +32,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
 /**
  * Deployment descriptor for a single input gate instance.
  *
- * <p> Each input gate consumes partitions of a single intermediate result. The consumed
+ * <p>Each input gate consumes partitions of a single intermediate result. The consumed
  * subpartition index is the same for each consumed partition.
  *
  * @see SingleInputGate

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/PartialInputChannelDeploymentDescriptor.java

@@ -31,7 +31,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
 /**
  * Partial deployment descriptor for a single input channel instance.
  *
- * <p> This deployment descriptor is created in {@link Execution#scheduleOrUpdateConsumers(java.util.List)},
+ * <p>This deployment descriptor is created in {@link Execution#scheduleOrUpdateConsumers(java.util.List)},
  * if the consumer instance is not yet clear. Once the instance on which the consumer runs is known,
  * the deployment descriptor is updated by completing the partition location.
  */

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

@@ -51,9 +51,9 @@ public class ResultPartitionDeploymentDescriptor implements Serializable {
 	/** The number of subpartitions. */
 	private final int numberOfSubpartitions;
 
-	/** The maximum parallelism */
+	/** The maximum parallelism. */
 	private final int maxParallelism;
-	
+
 	/** Flag whether the result partition should send scheduleOrUpdateConsumer messages. */
 	private final boolean sendScheduleOrUpdateConsumersMessage;
 

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionLocation.java

@@ -27,7 +27,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
 /**
  * Location of a result partition from the perspective of the consuming task.
  *
- * <p> The location indicates both the instance, on which the partition is produced and the state of
+ * <p>The location indicates both the instance, on which the partition is produced and the state of
  * the producing task. There are three possibilities:
  *
  * <ol>

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java

@@ -36,22 +36,22 @@ public final class TaskDeploymentDescriptor implements Serializable {
 
 	private static final long serialVersionUID = -3233562176034358530L;
 
-	/** Serialized job information */
+	/** Serialized job information. */
 	private final SerializedValue<JobInformation> serializedJobInformation;
 
-	/** Serialized task information */
+	/** Serialized task information. */
 	private final SerializedValue<TaskInformation> serializedTaskInformation;
 
 	/** The ID referencing the attempt to execute the task. */
 	private final ExecutionAttemptID executionId;
 
-	/** The allocation ID of the slot in which the task shall be run */
+	/** The allocation ID of the slot in which the task shall be run. */
 	private final AllocationID allocationId;
 
 	/** The task's index in the subtask group. */
 	private final int subtaskIndex;
 
-	/** Attempt number the task */
+	/** Attempt number the task. */
 	private final int attemptNumber;
 
 	/** The list of produced intermediate result partition deployment descriptors. */
@@ -60,10 +60,10 @@ public final class TaskDeploymentDescriptor implements Serializable {
 	/** The list of consumed intermediate result partitions. */
 	private final Collection<InputGateDeploymentDescriptor> inputGates;
 
-	/** Slot number to run the sub task in on the target machine */
+	/** Slot number to run the sub task in on the target machine. */
 	private final int targetSlotNumber;
 
-	/** State handles for the sub task */
+	/** State handles for the sub task. */
 	private final TaskStateHandles taskStateHandles;
 
 	public TaskDeploymentDescriptor(
@@ -130,7 +130,7 @@ public final class TaskDeploymentDescriptor implements Serializable {
 	}
 
 	/**
-	 * Returns the attempt number of the subtask
+	 * Returns the attempt number of the subtask.
 	 */
 	public int getAttemptNumber() {
 		return attemptNumber;

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptorTest.java

@@ -32,6 +32,7 @@ import org.apache.flink.runtime.io.network.ConnectionID;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
 import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
 import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
 import org.junit.Test;
 
 import java.net.InetAddress;
@@ -43,6 +44,9 @@ import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+/**
+ * Tests for the {@link InputChannelDeploymentDescriptor}.
+ */
 public class InputChannelDeploymentDescriptorTest {
 
 	/**
@@ -122,7 +126,6 @@ public class InputChannelDeploymentDescriptorTest {
 		ExecutionVertex consumer = mock(ExecutionVertex.class);
 		SimpleSlot consumerSlot = mockSlot(consumerResourceId);
 
-
 		// Unknown partition
 		ExecutionVertex unknownProducer = mockExecutionVertex(ExecutionState.CREATED, null); // no assigned resource
 		IntermediateResultPartition unknownPartition = mockPartition(unknownProducer);
@@ -143,7 +146,6 @@ public class InputChannelDeploymentDescriptorTest {
 		assertTrue(desc[0].getConsumedPartitionLocation().isUnknown());
 		assertNull(desc[0].getConsumedPartitionLocation().getConnectionId());
 
-
 		try {
 			// Fail if lazy deployment is *not* allowed
 			allowLazyDeployment = false;

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java

@@ -22,11 +22,15 @@ import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+/**
+ * Tests for the {@link ResultPartitionDeploymentDescriptor}.
+ */
 public class ResultPartitionDeploymentDescriptorTest {
 
 	/**

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java

@@ -18,31 +18,34 @@
 
 package org.apache.flink.runtime.deployment;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
-
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.List;
-
 import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.runtime.blob.BlobKey;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.executiongraph.JobInformation;
 import org.apache.flink.runtime.executiongraph.TaskInformation;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
 import org.apache.flink.runtime.operators.BatchTask;
-import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.runtime.state.TaskStateHandles;
 import org.apache.flink.util.SerializedValue;
 
 import org.junit.Test;
 
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link TaskDeploymentDescriptor}.
+ */
 public class TaskDeploymentDescriptorTest {
 	@Test
 	public void testSerialization() {
@@ -83,9 +86,9 @@ public class TaskDeploymentDescriptorTest {
 				taskStateHandles,
 				producedResults,
 				inputGates);
-	
+
 			final TaskDeploymentDescriptor copy = CommonTestUtils.createCopySerializable(orig);
-	
+
 			assertFalse(orig.getSerializedJobInformation() == copy.getSerializedJobInformation());
 			assertFalse(orig.getSerializedTaskInformation() == copy.getSerializedTaskInformation());
 			assertFalse(orig.getExecutionAttemptId() == copy.getExecutionAttemptId());