/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.streaming.runtime.metrics;

import org.apache.flink.metrics.Gauge;

/**
 * A {@link Gauge} for exposing the minimum watermark of a {@link WatermarkGauge} pair.
 */
public class MinWatermarkGauge implements Gauge<Long> {

	private WatermarkGauge watermarkGauge1;
	private WatermarkGauge watermarkGauge2;

	public MinWatermarkGauge(WatermarkGauge watermarkGauge1, WatermarkGauge watermarkGauge2) {
		this.watermarkGauge1 = watermarkGauge1;
		this.watermarkGauge2 = watermarkGauge2;
	}

	@Override
	public Long getValue() {
		return Math.min(watermarkGauge1.getValue(), watermarkGauge2.getValue());
	}
}

@@ -893,16 +893,108 @@
 Gauge;%0A%0A
+import java.util.Arrays;%0A%0Aimport static org.apache.flink.util.Preconditions.checkArgument;%0A%0A
 /**%0A * A
@@ -1046,17 +1046,22 @@
 mark of 
-a
+chosen
  %7B@link 
@@ -1079,13 +1079,9 @@
 uge%7D
- pair
+s
 .%0A *
@@ -1138,17 +1138,16 @@
 Long%3E %7B%0A
-%0A
 %09private
@@ -1151,48 +1151,13 @@
 ate 
-WatermarkGauge watermarkGauge1;%0A%09private
+final
  Wat
@@ -1159,32 +1159,34 @@
 l WatermarkGauge
+%5B%5D
  watermarkGauge2
@@ -1184,17 +1184,17 @@
 arkGauge
-2
+s
 ;%0A%0A%09publ
@@ -1233,40 +1233,11 @@
 uge 
-watermarkGauge1, WatermarkGauge 
+...
 wate
@@ -1250,38 +1250,29 @@
 auge
-2
+s
 ) %7B%0A%09%09
-this.watermarkGauge1 = 
+checkArgument(
 wate
@@ -1281,17 +1281,29 @@
 arkGauge
-1
+s.length %3E 0)
 ;%0A%09%09this
@@ -1317,17 +1317,17 @@
 arkGauge
-2
+s
  = water
@@ -1335,17 +1335,17 @@
 arkGauge
-2
+s
 ;%0A%09%7D%0A%0A%09@
@@ -1392,16 +1392,21 @@
 urn 
-Math.min
+Arrays.stream
 (wat
@@ -1420,23 +1420,26 @@
 auge
-1.getValue(), w
+s)%0A%09%09%09.mapToLong(W
 ater
@@ -1451,18 +1451,18 @@
 auge
-2.
+::
 getValue
 ());
@@ -1457,18 +1457,40 @@
 getValue
-(
 )
+%0A%09%09%09.min()%0A%09%09%09.orElse(0
 );%0A%09%7D%0A%7D%0A

Generalize MinWatermarkGauge to accept an aribtrary number of input gauges