HDDS-893. pipeline status is ALLOCATED in scmcli listPipelines command. Contributed by Lokesh Jain.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/cf571133 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/cf571133 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/cf571133 Branch: refs/heads/HDFS-12943 Commit: cf571133b89643e4db96ec4cc7988d4a2f850be9 Parents: b1ce9aa Author: Arpit Agarwal <[email protected]> Authored: Wed Dec 19 13:42:06 2018 +0530 Committer: Arpit Agarwal <[email protected]> Committed: Wed Dec 19 13:42:06 2018 +0530 ---------------------------------------------------------------------- .../common/helpers/ContainerWithPipeline.java | 7 ++- .../hadoop/hdds/scm/pipeline/Pipeline.java | 38 ++++++++++++++-- .../pipeline/UnknownPipelineStateException.java | 46 ++++++++++++++++++++ ...rLocationProtocolClientSideTranslatorPB.java | 9 ++-- ...rLocationProtocolServerSideTranslatorPB.java | 8 ++-- hadoop-hdds/common/src/main/proto/hdds.proto | 8 +++- .../hdds/scm/pipeline/SCMPipelineManager.java | 6 ++- 7 files changed, 107 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java index 8f49255..5b01bd2 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java @@ -24,6 +24,7 @@ import org.apache.commons.lang3.builder.HashCodeBuilder; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.container.ContainerInfo; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.UnknownPipelineStateException; /** * Class wraps ozone container info. @@ -48,13 +49,15 @@ public class ContainerWithPipeline implements Comparator<ContainerWithPipeline>, } public static ContainerWithPipeline fromProtobuf( - HddsProtos.ContainerWithPipeline allocatedContainer) { + HddsProtos.ContainerWithPipeline allocatedContainer) + throws UnknownPipelineStateException { return new ContainerWithPipeline( ContainerInfo.fromProtobuf(allocatedContainer.getContainerInfo()), Pipeline.getFromProtobuf(allocatedContainer.getPipeline())); } - public HddsProtos.ContainerWithPipeline getProtobuf() { + public HddsProtos.ContainerWithPipeline getProtobuf() + throws UnknownPipelineStateException { HddsProtos.ContainerWithPipeline.Builder builder = HddsProtos.ContainerWithPipeline.newBuilder(); builder.setContainerInfo(getContainerInfo().getProtobuf()) http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java index a103bd7..7fcfc8a 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java @@ -136,11 +136,13 @@ public final class Pipeline { return nodeStatus.isEmpty(); } - public HddsProtos.Pipeline getProtobufMessage() { + public HddsProtos.Pipeline getProtobufMessage() + throws UnknownPipelineStateException { HddsProtos.Pipeline.Builder builder = HddsProtos.Pipeline.newBuilder() .setId(id.getProtobuf()) .setType(type) .setFactor(factor) + .setState(PipelineState.getProtobuf(state)) .setLeaderID("") .addAllMembers(nodeStatus.keySet().stream() .map(DatanodeDetails::getProtoBufMessage) @@ -148,11 +150,13 @@ public final class Pipeline { return builder.build(); } - public static Pipeline getFromProtobuf(HddsProtos.Pipeline pipeline) { + public static Pipeline getFromProtobuf(HddsProtos.Pipeline pipeline) + throws UnknownPipelineStateException { + Preconditions.checkNotNull(pipeline, "Pipeline is null"); return new Builder().setId(PipelineID.getFromProtobuf(pipeline.getId())) .setFactor(pipeline.getFactor()) .setType(pipeline.getType()) - .setState(PipelineState.ALLOCATED) + .setState(PipelineState.fromProtobuf(pipeline.getState())) .setNodes(pipeline.getMembersList().stream() .map(DatanodeDetails::getFromProtoBuf).collect(Collectors.toList())) .build(); @@ -270,6 +274,32 @@ public final class Pipeline { * Possible Pipeline states in SCM. */ public enum PipelineState { - ALLOCATED, OPEN, CLOSED + ALLOCATED, OPEN, CLOSED; + + public static PipelineState fromProtobuf(HddsProtos.PipelineState state) + throws UnknownPipelineStateException { + Preconditions.checkNotNull(state, "Pipeline state is null"); + switch (state) { + case PIPELINE_ALLOCATED: return ALLOCATED; + case PIPELINE_OPEN: return OPEN; + case PIPELINE_CLOSED: return CLOSED; + default: + throw new UnknownPipelineStateException( + "Pipeline state: " + state + " is not recognized."); + } + } + + public static HddsProtos.PipelineState getProtobuf(PipelineState state) + throws UnknownPipelineStateException { + Preconditions.checkNotNull(state, "Pipeline state is null"); + switch (state) { + case ALLOCATED: return HddsProtos.PipelineState.PIPELINE_ALLOCATED; + case OPEN: return HddsProtos.PipelineState.PIPELINE_OPEN; + case CLOSED: return HddsProtos.PipelineState.PIPELINE_CLOSED; + default: + throw new UnknownPipelineStateException( + "Pipeline state: " + state + " is not recognized."); + } + } } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/UnknownPipelineStateException.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/UnknownPipelineStateException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/UnknownPipelineStateException.java new file mode 100644 index 0000000..1aa9358 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/UnknownPipelineStateException.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.scm.pipeline; + + import java.io.IOException; + +/** + * Signals that a pipeline state is not recognized. + */ +public class UnknownPipelineStateException extends IOException { + /** + * Constructs an {@code UnknownPipelineStateException} with {@code null} + * as its error detail message. + */ + public UnknownPipelineStateException() { + super(); + } + + /** + * Constructs an {@code UnknownPipelineStateException} with the specified + * detail message. + * + * @param message + * The detail message (which is saved for later retrieval + * by the {@link #getMessage()} method) + */ + public UnknownPipelineStateException(String message) { + super(message); + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java index 117e58d..a24ba88 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java @@ -315,9 +315,12 @@ public final class StorageContainerLocationProtocolClientSideTranslatorPB .newBuilder().build(); ListPipelineResponseProto response = rpcProxy.listPipelines( NULL_RPC_CONTROLLER, request); - return response.getPipelinesList().stream() - .map(Pipeline::getFromProtobuf) - .collect(Collectors.toList()); + List<Pipeline> list = new ArrayList<>(); + for (HddsProtos.Pipeline pipeline : response.getPipelinesList()) { + Pipeline fromProtobuf = Pipeline.getFromProtobuf(pipeline); + list.add(fromProtobuf); + } + return list; } catch (ServiceException e) { throw ProtobufHelper.getRemoteException(e); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java index 2ae559a..1630875 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java @@ -227,9 +227,11 @@ public final class StorageContainerLocationProtocolServerSideTranslatorPB try { ListPipelineResponseProto.Builder builder = ListPipelineResponseProto .newBuilder(); - List<Pipeline> pipelineIDs = impl.listPipelines(); - pipelineIDs.stream().map(Pipeline::getProtobufMessage) - .forEach(builder::addPipelines); + List<Pipeline> pipelines = impl.listPipelines(); + for (Pipeline pipeline : pipelines) { + HddsProtos.Pipeline protobufMessage = pipeline.getProtobufMessage(); + builder.addPipelines(protobufMessage); + } return builder.build(); } catch (IOException e) { throw new ServiceException(e); http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/common/src/main/proto/hdds.proto ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/proto/hdds.proto b/hadoop-hdds/common/src/main/proto/hdds.proto index cf3d6d4..2d8f43f 100644 --- a/hadoop-hdds/common/src/main/proto/hdds.proto +++ b/hadoop-hdds/common/src/main/proto/hdds.proto @@ -44,11 +44,17 @@ message PipelineID { required string id = 1; } +enum PipelineState { + PIPELINE_ALLOCATED = 1; + PIPELINE_OPEN = 2; + PIPELINE_CLOSED = 3; +} + message Pipeline { required string leaderID = 1; repeated DatanodeDetailsProto members = 2; // TODO: remove the state and leaderID from this class - optional LifeCycleState state = 3 [default = OPEN]; + optional PipelineState state = 3 [default = PIPELINE_ALLOCATED]; optional ReplicationType type = 4 [default = STAND_ALONE]; optional ReplicationFactor factor = 5 [default = ONE]; required PipelineID id = 6; http://git-wip-us.apache.org/repos/asf/hadoop/blob/cf571133/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java index 382483f..8b7a56d 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java @@ -99,8 +99,10 @@ public class SCMPipelineManager implements PipelineManager { (MetadataKeyFilters.MetadataKeyFilter[])null); for (Map.Entry<byte[], byte[]> entry : pipelines) { - Pipeline pipeline = Pipeline.getFromProtobuf( - HddsProtos.Pipeline.PARSER.parseFrom(entry.getValue())); + HddsProtos.Pipeline.Builder pipelineBuilder = HddsProtos.Pipeline + .newBuilder(HddsProtos.Pipeline.PARSER.parseFrom(entry.getValue())); + Pipeline pipeline = Pipeline.getFromProtobuf(pipelineBuilder.setState( + HddsProtos.PipelineState.PIPELINE_ALLOCATED).build()); Preconditions.checkNotNull(pipeline); stateManager.addPipeline(pipeline); nodeManager.addPipeline(pipeline); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
