[hotfix][kinesis] Update emit record javadoc and don't count max wate…

…rmark as timeout
nicusX · Aug 27, 2019 · b6bd8f6 · b6bd8f6
1 parent 0437ad2
commit b6bd8f6
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/...main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java b/...main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
@@ -715,7 +715,7 @@ protected KinesisDeserializationSchema<T> getClonedDeserializationSchema() {
 	// ------------------------------------------------------------------------
 
 	/**
-	 * Atomic operation to collect a record and update state to the sequence number of the record.
+	 * Prepare a record and hand it over to the {@link RecordEmitter}, which may collect it asynchronously.
 	 * This method is called by {@link ShardConsumer}s.
 	 *
 	 * @param record the record to collect
@@ -752,7 +752,8 @@ protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shar
 	}
 
 	/**
-	 * Actual record emission called from the record emitter.
+	 * Atomic operation to collect a record and update state to the sequence number of the record.
+	 * This method is called from the record emitter.
 	 *
 	 * <p>Responsible for tracking per shard watermarks and emit timestamps extracted from
 	 * the record, when a watermark assigner was configured.

diff --git a/...n/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTracker.java b/...n/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTracker.java
@@ -144,7 +144,9 @@ public byte[] getResult(Map<String, WatermarkState> accumulator) {
 				WatermarkState ws = e.getValue();
 				if (ws.lastUpdated + updateTimeoutMillis < currentTime) {
 					// ignore outdated entry
-					updateTimeoutCount++;
+					if (ws.watermark < Long.MAX_VALUE) {
+						updateTimeoutCount++;
+					}
 					continue;
 				}
 				globalWatermark = Math.min(ws.watermark, globalWatermark);