Revert "[FLINK-10727][network] remove unnecessary synchronization in …

…SingleInputGate#requestPartitions()"

This reverts commit 194603a because it caused the
creation of sub partitions to fail.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

@@ -477,28 +477,26 @@ public boolean isFinished() {
 
  @Override
  public void requestPartitions() throws IOException, InterruptedException {
- if (requestedPartitionsFlag) {
- return;
- }
-
  synchronized (requestLock) {
- if (isReleased) {
- throw new IllegalStateException("Already released.");
- }
+ if (!requestedPartitionsFlag) {
+ if (isReleased) {
+ throw new IllegalStateException("Already released.");
+ }
 
- // Sanity checks
- if (numberOfInputChannels != inputChannels.size()) {
- throw new IllegalStateException("Bug in input gate setup logic: mismatch between" +
- "number of total input channels and the currently set number of input " +
- "channels.");
- }
+  // Sanity checks
+  if (numberOfInputChannels != inputChannels.size()) {
+  throw new IllegalStateException("Bug in input gate setup logic: mismatch between" +
+  "number of total input channels and the currently set number of input " +
+  "channels.");
+  }
 
- for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition(consumedSubpartitionIndex);
+ for (InputChannel inputChannel : inputChannels.values()) {
+ inputChannel.requestSubpartition(consumedSubpartitionIndex);
+ }
  }
- }
 
- requestedPartitionsFlag = true;
+ requestedPartitionsFlag = true;
+ }
  }
 
  // ------------------------------------------------------------------------