From b2363be853dc10957e235ef9fb819621b40c5d6b Mon Sep 17 00:00:00 2001 From: guohao1 Date: Mon, 20 Mar 2023 16:28:20 +0800 Subject: [PATCH 1/4] HDDS-8220. [Ozone-Streaming] KeyValueStreamDataChannel add write data exception handling --- .../impl/KeyValueStreamDataChannel.java | 22 ++++++++++++++++--- .../keyvalue/impl/StreamDataChannelBase.java | 4 ++++ 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java index 99dc40f5d002..3bc3e2bf9cc7 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java @@ -47,6 +47,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import static org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil.onFailure; + /** * This class is used to get the DataChannel for streaming. */ @@ -158,7 +160,12 @@ ContainerProtos.Type getType() { public int write(ReferenceCountedObject referenceCounted) throws IOException { assertOpen(); - return writeBuffers(referenceCounted, buffers, super::writeFileChannel); + try { + return writeBuffers(referenceCounted, buffers, super::writeFileChannel); + } catch (IOException e) { + volumeOnFailure(); + throw e; + } } static int writeBuffers(ReferenceCountedObject src, @@ -198,8 +205,13 @@ void assertOpen() throws IOException { @Override public void close() throws IOException { if (closed.compareAndSet(false, true)) { - putBlockRequest.set(closeBuffers(buffers, super::writeFileChannel)); - super.close(); + try { + putBlockRequest.set(closeBuffers(buffers, super::writeFileChannel)); + super.close(); + } catch (IOException e) { + volumeOnFailure(); + throw e; + } } } @@ -274,4 +286,8 @@ private static ContainerCommandRequestProto readPutBlockRequest(ByteBuffer b) } return request; } + + private void volumeOnFailure() { + onFailure(getContainerData().getVolume()); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java index 982903324848..9e0919589a4d 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java @@ -64,6 +64,10 @@ private FileChannel getChannel() { return randomAccessFile.getChannel(); } + public ContainerData getContainerData() { + return containerData; + } + @Override public final void force(boolean metadata) throws IOException { getChannel().force(metadata); From 8a7b5e98d98ec4191b4fbf2ee06f2b46933f7511 Mon Sep 17 00:00:00 2001 From: guohao1 Date: Mon, 20 Mar 2023 22:01:48 +0800 Subject: [PATCH 2/4] trigger new CI From b043546bcd1263ddeaf9261ff7d25d6f514986ec Mon Sep 17 00:00:00 2001 From: guohao1 Date: Tue, 21 Mar 2023 10:09:34 +0800 Subject: [PATCH 3/4] rename method --- .../impl/KeyValueStreamDataChannel.java | 21 ++---------- .../keyvalue/impl/StreamDataChannelBase.java | 33 ++++++++++++++----- 2 files changed, 28 insertions(+), 26 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java index 3bc3e2bf9cc7..5b5035466ae1 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java @@ -47,8 +47,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import static org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil.onFailure; - /** * This class is used to get the DataChannel for streaming. */ @@ -160,12 +158,7 @@ ContainerProtos.Type getType() { public int write(ReferenceCountedObject referenceCounted) throws IOException { assertOpen(); - try { - return writeBuffers(referenceCounted, buffers, super::writeFileChannel); - } catch (IOException e) { - volumeOnFailure(); - throw e; - } + return writeBuffers(referenceCounted, buffers, super::writeFileChannel); } static int writeBuffers(ReferenceCountedObject src, @@ -205,13 +198,8 @@ void assertOpen() throws IOException { @Override public void close() throws IOException { if (closed.compareAndSet(false, true)) { - try { - putBlockRequest.set(closeBuffers(buffers, super::writeFileChannel)); - super.close(); - } catch (IOException e) { - volumeOnFailure(); - throw e; - } + putBlockRequest.set(closeBuffers(buffers, super::writeFileChannel)); + super.close(); } } @@ -287,7 +275,4 @@ private static ContainerCommandRequestProto readPutBlockRequest(ByteBuffer b) return request; } - private void volumeOnFailure() { - onFailure(getContainerData().getVolume()); - } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java index 9e0919589a4d..4b1a255e93ef 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java @@ -31,6 +31,8 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; +import static org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil.onFailure; + /** * For write state machine data. */ @@ -64,13 +66,18 @@ private FileChannel getChannel() { return randomAccessFile.getChannel(); } - public ContainerData getContainerData() { - return containerData; + protected void checkVolume() { + onFailure(containerData.getVolume()); } @Override public final void force(boolean metadata) throws IOException { - getChannel().force(metadata); + try { + getChannel().force(metadata); + } catch (IOException e) { + checkVolume(); + throw e; + } } @Override @@ -80,14 +87,24 @@ public final boolean isOpen() { @Override public void close() throws IOException { - randomAccessFile.close(); + try { + randomAccessFile.close(); + } catch (IOException e) { + checkVolume(); + throw e; + } } final int writeFileChannel(ByteBuffer src) throws IOException { - final int writeBytes = getChannel().write(src); - metrics.incContainerBytesStats(getType(), writeBytes); - containerData.updateWriteStats(writeBytes, false); - return writeBytes; + try { + final int writeBytes = getChannel().write(src); + metrics.incContainerBytesStats(getType(), writeBytes); + containerData.updateWriteStats(writeBytes, false); + return writeBytes; + } catch (IOException e) { + checkVolume(); + throw e; + } } @Override From a9824aaa654478fcade0fcc38cd68feb3fcece15 Mon Sep 17 00:00:00 2001 From: guohao1 Date: Tue, 21 Mar 2023 10:13:12 +0800 Subject: [PATCH 4/4] unused code --- .../ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java index 5b5035466ae1..99dc40f5d002 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java @@ -274,5 +274,4 @@ private static ContainerCommandRequestProto readPutBlockRequest(ByteBuffer b) } return request; } - }