[fix](broker) fix export job failed for that currentStreamOffset may …

…be different with request offset (#23133)

Co-authored-by: caiconghui1 <[email protected]>when export job encounter heavy pressure, the failed export job may see the following message
current outputstream offset is 423597 not equal to request 421590, cause by: null,
because the broker pwrite operation may retry for timeout, so we just skip it instead of throw broker exception.

fs_brokers/apache_hdfs_broker/src/main/java/org/apache/doris/broker/hdfs/FileSystemManager.java

@@ -1279,16 +1279,24 @@ public void pwrite(TBrokerFD fd, long offset, byte[] data) {
                         "errors while get file pos from output stream");
             }
             if (currentStreamOffset != offset) {
-                throw new BrokerException(TBrokerOperationStatusCode.INVALID_INPUT_OFFSET,
+                // it's ok, it means that last pwrite succeed finally
+                if (currentStreamOffset == offset + data.length) {
+                    logger.warn("invalid offset, but current outputstream offset is "
+                        + currentStreamOffset + ", data length is " + data.length + ", the sum is equal to request offset "
+                        + offset + ", so just skip it");
+                } else {
+                    throw new BrokerException(TBrokerOperationStatusCode.INVALID_INPUT_OFFSET,
                         "current outputstream offset is {} not equal to request {}",
                         currentStreamOffset, offset);
-            }
-            try {
-                fsDataOutputStream.write(data);
-            } catch (IOException e) {
-                logger.error("errors while write data to output stream", e);
-                throw new BrokerException(TBrokerOperationStatusCode.TARGET_STORAGE_SERVICE_ERROR,
+                }
+            } else {
+                try {
+                    fsDataOutputStream.write(data);
+                } catch (IOException e) {
+                    logger.error("errors while write data to output stream", e);
+                    throw new BrokerException(TBrokerOperationStatusCode.TARGET_STORAGE_SERVICE_ERROR,
                         e, "errors while write data to output stream");
+                }
             }
         }
     }