flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] zhijiangW commented on a change in pull request #6705: [FLINK-10356][network] add sanity checks to SpillingAdaptiveSpanningRecordDeserializer
Date Tue, 16 Oct 2018 04:51:29 GMT
zhijiangW commented on a change in pull request #6705: [FLINK-10356][network] add sanity checks
to SpillingAdaptiveSpanningRecordDeserializer
URL: https://github.com/apache/flink/pull/6705#discussion_r225396776
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
 ##########
 @@ -549,21 +584,53 @@ private void addNextChunkFromMemorySegment(MemorySegment segment, int
offset, in
 				}
 				else {
 					spillingChannel.close();
+					spillingChannel = null;
 
-					BufferedInputStream inStream = new BufferedInputStream(new FileInputStream(spillFile),
2 * 1024 * 1024);
+					BufferedInputStream inStream =
+						new BufferedInputStream(
+							new FileInputStream(checkNotNull(spillFile)),
+							2 * 1024 * 1024);
 					this.spillFileReader = new DataInputViewStreamWrapper(inStream);
 				}
 			}
 		}
 
-		private void moveRemainderToNonSpanningDeserializer(NonSpanningWrapper deserializer) {
+		private void moveRemainderToNonSpanningDeserializer(NonSpanningWrapper deserializer) throws
IOException {
+			Optional<String> deserializationError = getDeserializationError(0);
+			if (deserializationError.isPresent()) {
+				throw new IOException(deserializationError.get());
+			}
+
 			deserializer.clear();
 
 			if (leftOverData != null) {
 				deserializer.initializeFromMemorySegment(leftOverData, leftOverStart, leftOverLimit);
 			}
 		}
 
+		private Optional<String> getDeserializationError(int addToReadBytes) {
+			Optional<String> deserializationError = Optional.empty();
+			int remainingSpanningBytes = 0, leftOverDataStart = 0, leftOverDataLimit = 0;
+			if (this.spillFileReader == null) {
+				remainingSpanningBytes = this.serializationReadBuffer.available() - addToReadBytes;
+			} else {
+				try {
+					remainingSpanningBytes = this.spillFileReader.available() - addToReadBytes;
+				} catch (IOException ignored) {
+				}
+			}
+			if (this.leftOverData != null) {
 
 Review comment:
   Because the following condition `remainingSpanningBytes != 0` is not always true, this
condition is not always necessary. I am not sure whether it is worth putting this condition
inside the below one.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message