fix(client): wait for all asynchronous flush requests to complete

close:#1000511680

Signed-off-by: chihe <chihe@oppo.com>

sdk/data/stream/extent_cache.go

@@ -26,11 +26,10 @@ import (
 
 // ExtentRequest defines the struct for the request of read or write an extent.
 type ExtentRequest struct {
-	FileOffset  int
-	Size        int
-	Data        []byte
-	ExtentKey   *proto.ExtentKey
-	CreateNewEk bool
+	FileOffset int
+	Size       int
+	Data       []byte
+	ExtentKey  *proto.ExtentKey
 }
 
 // String returns the string format of the extent request.
@@ -533,7 +532,6 @@ func (cache *ExtentCache) PrepareWriteRequests(offset, size int, data []byte) []
 		log.LogDebugf("action[ExtentCache.PrepareWriteRequests] ek [%v], pivot[%v]", ek, pivot)
 		return false
 	})
-	createNewExtentKey := false
 	cache.root.AscendRange(lower, upper, func(i btree.Item) bool {
 		ek := i.(*proto.ExtentKey)
 		ekStart := int(ek.FileOffset)
@@ -557,10 +555,7 @@ func (cache *ExtentCache) PrepareWriteRequests(offset, size int, data []byte) []
 				start = end
 				return false
 			} else {
-				// exact cover current extent: start == ekStart && end == ekEnd
-				if start == ekStart && end == ekEnd {
-					createNewExtentKey = true
-				}
+				// create new extentKey if end >= ekEnd
 				return true
 			}
 		} else if start < ekEnd {
@@ -586,9 +581,6 @@ func (cache *ExtentCache) PrepareWriteRequests(offset, size int, data []byte) []
 	if start < end {
 		// add hole (start, end)
 		req := NewExtentRequest(start, end-start, data[start-offset:end-offset], nil)
-		if createNewExtentKey {
-			req.CreateNewEk = true
-		}
 		requests = append(requests, req)
 	}
 

sdk/data/stream/stream_writer.go

@@ -385,7 +385,9 @@ begin:
 	isChecked := false
 	// Must flush before doing overwrite
 	for _, req := range requests {
-		if !s.waitForFlush && req.ExtentKey == nil && !req.CreateNewEk {
+		// if offset < fileSize, client should wait for all the async flush requests
+		// to complete
+		if !s.waitForFlush && req.ExtentKey == nil && offset >= fileSize {
 			continue
 		}
 		err = s.flush(true, uuid.New().String())