[hotfix] Minor edits to comments in code

.travis.yml

@@ -13,7 +13,7 @@ language: java
 
 #See https://issues.apache.org/jira/browse/FLINK-1072
 # NOTE: When changing the build matrix, please also check with the tools/deploy_to_maven.sh file
-# The file assumes a certain build order for the maven / nigthly build deployments.
+# The file assumes a certain build order for the maven / nightly build deployments.
 matrix:
  include:
  - jdk: "oraclejdk8"

flink-core/src/main/java/org/apache/flink/api/common/Plan.java

@@ -288,7 +288,7 @@ public int getDefaultParallelism() {
  */
  public void setDefaultParallelism(int defaultParallelism) {
  checkArgument(defaultParallelism >= 1 || defaultParallelism == -1,
- "The default parallelism must be positive, or -1 if the system should use the globally comfigured default.");
+ "The default parallelism must be positive, or -1 if the system should use the globally configured default.");
 
  this.defaultParallelism = defaultParallelism;
  }

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java

@@ -78,7 +78,7 @@ public int getArity() {
  }
 
  // We use this in the Aggregate and Distinct Operators to create instances
- // of immutable Typles (i.e. Scala Tuples)
+ // of immutable Tuples (i.e. Scala Tuples)
  public abstract T createInstance(Object[] fields);
 
  public abstract T createOrReuseInstance(Object[] fields, T reuse);

flink-dist/src/main/flink-bin/bin/config.sh

@@ -271,7 +271,7 @@ if [ -n "${HBASE_CONF_DIR}" ]; then
 fi
 
 # Auxilliary function which extracts the name of host from a line which
-# also potentialy includes topology information and the taskManager type
+# also potentially includes topology information and the taskManager type
 extractHostName() {
  # handle comments: extract first part of string (before first # character)
  SLAVE=`echo $1 | cut -d'#' -f 1`

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/EdgeJoinFunction.java

@@ -21,6 +21,7 @@
 import java.io.Serializable;
 
 import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.java.DataSet;
 
 /**
  * Interface to be implemented by the transformation function

flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java

@@ -473,12 +473,12 @@ else if (graphCreator.getSinks().size() > 1) {
  }
 
  // now that we have all nodes created and recorded which ones consume memory, tell the nodes their minimal
- // guaranteed memory, for further cost estimations. we assume an equal distribution of memory among consumer tasks
+ // guaranteed memory, for further cost estimations. We assume an equal distribution of memory among consumer tasks
  rootNode.accept(new IdAndEstimatesVisitor(this.statistics));
 
  // We are dealing with operator DAGs, rather than operator trees.
  // That requires us to deviate at some points from the classical DB optimizer algorithms.
- // This step build some auxiliary structures to help track branches and joins in the DAG
+ // This step builds auxiliary structures to help track branches and joins in the DAG
  BranchesVisitor branchingVisitor = new BranchesVisitor();
  rootNode.accept(branchingVisitor);
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FilterNode.java

@@ -29,7 +29,7 @@
 import org.apache.flink.optimizer.operators.OperatorDescriptorSingle;
 
 /**
- * The optimizer's internal representation of a <i>FlatMap</i> operator node.
+ * The optimizer's internal representation of a <i>Filter</i> operator node.
  */
 public class FilterNode extends SingleInputNode {
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/ChannelWriterOutputView.java

@@ -29,7 +29,7 @@
 /**
  * A {@link org.apache.flink.core.memory.DataOutputView} that is backed by a
  * {@link BlockChannelWriter}, making it effectively a data output
- * stream. The view writes it data in blocks to the underlying channel, adding a minimal header to each block.
+ * stream. The view writes its data in blocks to the underlying channel, adding a minimal header to each block.
  * The data can be re-read by a {@link ChannelReaderInputView}, if it uses the same block size.
  */
 public final class ChannelWriterOutputView extends AbstractPagedOutputView {
@@ -116,7 +116,7 @@ public ChannelWriterOutputView(BlockChannelWriter<MemorySegment> writer, List<Me
 
  /**
  * Creates an new ChannelWriterOutputView that writes to the given channel. It uses only a single
- * memory segment for the buffering, which it takes from the writers return queue.
+ * memory segment for the buffering, which it takes from the writer's return queue.
  * Note that this variant locks if no buffers are contained
  * in the return queue.
  * 

flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java

@@ -55,7 +55,7 @@ public void setup(TaskContext<MapPartitionFunction<IT, OT>, OT> context) {
  }
 
  @Override
-  public int getNumberOfInputs() {
+ public int getNumberOfInputs() {
  return 1;
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java

@@ -1749,10 +1749,11 @@ private void init() {
  }
 
  public BT next(BT reuse) {
- // search unprobed record in bucket, while found none, move to next bucket and search.
+ // search unprobed record in bucket, while none found move to next bucket and search.
  while (true) {
  BT result = nextInBucket(reuse);
  if (result == null) {
+ // return null while there are no more buckets.
  if (!moveToNextOnHeapBucket()) {
  return null;
  }
@@ -1763,11 +1764,11 @@ public BT next(BT reuse) {
  }
 
  public BT next() {
- // search unProbed record in bucket, while found none, move to next bucket and search.
+ // search unprobed record in bucket, while none found move to next bucket and search.
  while (true) {
  BT result = nextInBucket();
  if (result == null) {
- // return null while there is no more bucket.
+ // return null while there are no more buckets.
  if (!moveToNextOnHeapBucket()) {
  return null;
  }

flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NormalizedKeySorter.java

@@ -361,7 +361,7 @@ public int compare(int i, int j) {
  }
 
  final long pointerI = segI.getLong(segmentOffsetI) & POINTER_MASK;
- final long pointerJ = segJ.getLong(segmentOffsetJ)  & POINTER_MASK;
+ final long pointerJ = segJ.getLong(segmentOffsetJ) & POINTER_MASK;
 
  return compareRecords(pointerI, pointerJ);
  }