This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new ae7110eaf2 HDDS-5852. Optimize streaming chunk write on exception
(#4839)
ae7110eaf2 is described below
commit ae7110eaf22b4c380f34467eb78231c881dfd40a
Author: Sadanand Shenoy <[email protected]>
AuthorDate: Mon Jan 15 18:19:07 2024 +0530
HDDS-5852. Optimize streaming chunk write on exception (#4839)
---
.../hadoop/hdds/scm/storage/BlockDataStreamOutput.java | 4 ++--
.../hadoop/ozone/client/io/KeyDataStreamOutput.java | 18 ++++++++++++++++++
2 files changed, 20 insertions(+), 2 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
index f878659afa..005402efa7 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
@@ -377,7 +377,7 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
* @return minimum commit index replicated to all nodes
* @throws IOException IOException in case watch gets timed out
*/
- private void watchForCommit(boolean bufferFull) throws IOException {
+ public void watchForCommit(boolean bufferFull) throws IOException {
checkOpen();
try {
XceiverClientReply reply = bufferFull ?
@@ -405,7 +405,7 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
* @param force true if no data was written since most recent putBlock and
* stream is being closed
*/
- private void executePutBlock(boolean close,
+ public void executePutBlock(boolean close,
boolean force) throws IOException {
checkOpen();
long flushPos = totalDataFlushedLength;
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java
index 2368cd78e9..e5a43819a3 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java
@@ -30,6 +30,7 @@ import
org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.storage.AbstractDataStreamOutput;
+import org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
@@ -260,6 +261,23 @@ public class KeyDataStreamOutput extends
AbstractDataStreamOutput
long containerId = streamEntry.getBlockID().getContainerID();
Collection<DatanodeDetails> failedServers = streamEntry.getFailedServers();
Preconditions.checkNotNull(failedServers);
+ if (!containerExclusionException) {
+ BlockDataStreamOutputEntry currentStreamEntry =
+ blockDataStreamOutputEntryPool.getCurrentStreamEntry();
+ if (currentStreamEntry != null) {
+ try {
+ BlockDataStreamOutput blockDataStreamOutput =
+ (BlockDataStreamOutput) currentStreamEntry
+ .getByteBufStreamOutput();
+ blockDataStreamOutput.executePutBlock(false, false);
+ blockDataStreamOutput.watchForCommit(false);
+ } catch (IOException e) {
+ LOG.error(
+ "Failed to execute putBlock/watchForCommit. " +
+ "Continuing to write chunks" + "in new block", e);
+ }
+ }
+ }
ExcludeList excludeList = blockDataStreamOutputEntryPool.getExcludeList();
long bufferedDataLen = blockDataStreamOutputEntryPool.computeBufferData();
if (!failedServers.isEmpty()) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
