[Problem] Inconsistent blocks when reading shuffle data

[zuston]

I found some tasks of spark jobs will throw the exceptions that the inconsistent blocks number. The stacktrace is as follows

22/09/03 15:29:21 ERROR Executor: Exception in task 330.0 in stage 9.0 (TID 59001)
org.apache.uniffle.common.exception.RssException: Blocks read inconsistent: expected 30000 blocks, actual 15636 blocks
	at org.apache.uniffle.client.impl.ShuffleReadClientImpl.checkProcessedBlockIds(ShuffleReadClientImpl.java:215)
	at org.apache.spark.shuffle.reader.RssShuffleDataIterator.hasNext(RssShuffleDataIterator.java:135)
	at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)

I didn't find any error/warn log in shuffle server which stored the corresponding partition data.

We dont set any replica config and directly use the MEMORY_LOCALFILE storageType. Does this exception caused by the disk error?

[roryqi]

It may be a bug of our rss. I ever try to fix the similar problems by #40 and Tencent/Firestorm#92. Could you reproduce this problem? Are the missing blocks disk storage or memory storage?

[zuston]

Could you reproduce this problem?

It occurs in our users' jobs and maybe I have to write test spark code to reproduce

Are the missing blocks disk storage or memory storage?

I dont know which storage's block missing. But I check the request number in shuffle server and the request log entry number in tasks, they are equal.

[zuston]

It may be a bug of our rss. I ever try to fix the similar problems by #40 and Tencent/Firestorm#92

It looks this problem is not the same as the problem you said. This problem is caused by lacking partial blocks.

[smallzhongfeng]

22/09/05 20:58:01 INFO impl.HdfsShuffleReadHandler: Read index files hdfs://ns1/tmp/4/308-308/xxxx-19999_0.index for 2 ms
22/09/05 20:58:01 WARN util.RssUtils: Read index data under flow
java.nio.BufferUnderflowException
	at java.nio.Buffer.nextGetIndex(Buffer.java:506)
	at java.nio.HeapByteBuffer.getLong(HeapByteBuffer.java:412)
	at org.apache.uniffle.common.util.RssUtils.transIndexDataToSegments(RssUtils.java:204)
	at org.apache.uniffle.common.util.RssUtils.transIndexDataToSegments(RssUtils.java:188)
	at org.apache.uniffle.storage.handler.impl.DataSkippableReadHandler.readShuffleData(DataSkippableReadHandler.java:67)
	at org.apache.uniffle.storage.handler.impl.HdfsClientReadHandler.readShuffleData(HdfsClientReadHandler.java:135)
	at org.apache.uniffle.storage.handler.impl.ComposedClientReadHandler.readShuffleData(ComposedClientReadHandler.java:112)
	at org.apache.uniffle.storage.handler.impl.ComposedClientReadHandler.readShuffleData(ComposedClientReadHandler.java:140)
	at org.apache.uniffle.client.impl.ShuffleReadClientImpl.read(ShuffleReadClientImpl.java:195)
	at org.apache.uniffle.client.impl.ShuffleReadClientImpl.readShuffleBlockData(ShuffleReadClientImpl.java:131)
	at org.apache.spark.shuffle.reader.RssShuffleDataIterator.hasNext(RssShuffleDataIterator.java:100)
	at org.apache.spark.shuffle.reader.RssShuffleReader$MultiPartitionIterator.hasNext(RssShuffleReader.java:213)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.sort_addToSorter_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown Source)

We seem to have the same problem, but the reason for the inconsistent blocks number is that an exception occurred when the index file was read.

[roryqi]

transIndexDataToSegments

It seems that we read incomplete index file.

[zuston]

transIndexDataToSegments

It seems that we read incomplete index file.

We should fail fast when encountering BufferUnderflowException instead of ignoring. https://github.com/apache/incubator-uniffle/blob/07f70ed872b87107fc7028577bd9e66d8349fd6c/common/src/main/java/org/apache/uniffle/common/util/RssUtils.java#L224

[roryqi]

For MEMORY_HDFS or MEMORY_LOCALFILE, it's normal to read incomplete index data, so we choose to ignore instead of failing fast.

[smallzhongfeng]

We should fail fast when encountering BufferUnderflowException instead of ignoring.

What I am curious about is whether this is caused by the fact that the data of the stream is not flushed normally or the stream is closed abnormally.

[roryqi]

We should fail fast when encountering BufferUnderflowException instead of ignoring.

What I am curious about is whether this is caused by the fact that the data of the stream is not flushed normally or the stream is closed abnormally.

Maybe shuffle index is writing..

[zuston]

For MEMORY_HDFS or MEMORY_LOCALFILE, it's normal to read incomplete index data, so we choose to ignore instead of failing fast.

Oh, make sense.

We should fail fast when encountering BufferUnderflowException instead of ignoring.

What I am curious about is whether this is caused by the fact that the data of the stream is not flushed normally or the stream is closed abnormally.

Maybe shuffle index is writing..

Firstly, maybe we should reproduce this exception that make reader/writer operate this file at the same time.

[zuston]

Revisit: do we need to fail fast?

As I know, the exception of BufferUnderflowException is caused by the incomplete FileBasedShuffleSegment. In localfile reader, it will always get the n pieces of FileBasedShuffleSegment and then wont throw the exception. However in HDFS index read, it may return the incomplete data ((this problem will happen when spill data from memory to HDFS, so we also should do similar operation like local index file read). After that, we should do fail fast in https://github.com/apache/incubator-uniffle/blob/07f70ed872b87107fc7028577bd9e66d8349fd6c/common/src/main/java/org/apache/uniffle/common/util/RssUtils.java#L224

Is inconsistent blocks caused by this BufferUnderflowException?

The operation of spilling is a common action for uniffle and it has done the operation of filtering blocks. So this inconsistent blocks means it exist bug when spilling

[leixm]

Our production environment also had such an exception once, but so far it has not been able to reproduce.

[leixm]

I think maybe the shuffle server is writing this file and the output stream has not been closed.

[leixm]

Can you help me understand #issue_63? @jerqi
When we use the storageType MEMORY_HDFS, MEMORY_LOCALFILE_HDFS, after we read data from memory, the data may be flushed to HDFS. So we could read incomplete index data.
org.apache.uniffle.storage.handler.impl.HdfsFileWriter#writeIndex is a complete operation, it looks like it shouldn't write an incomplete index file.

[leixm]

In our production environment, certain tasks will have this problem for a certain period of time. I am trying to reproduce this situation and keep the index file and data file.