You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2018/10/05 12:56:28 UTC
[33/51] [partial] hadoop git commit: HADOOP-15791. Remove Ozone
related sources from the 3.2 branch. Contributed by Elek, Marton.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
deleted file mode 100644
index 503112d..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
+++ /dev/null
@@ -1,236 +0,0 @@
-/*
- * 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.ozone.container.ozoneimpl;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.PipelineID;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
-import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
-import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
-import org.apache.hadoop.ozone.container.common.impl.HddsDispatcher;
-import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
-import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
-import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerGrpc;
-import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi;
-import org.apache.hadoop.ozone.container.common.transport.server.ratis.XceiverServerRatis;
-import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
-import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
-
-import org.apache.hadoop.ozone.container.replication.GrpcReplicationService;
-import org.apache.hadoop.ozone.container.replication
- .OnDemandContainerReplicationSource;
-import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.*;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-import static org.apache.hadoop.ozone.OzoneConsts.INVALID_PORT;
-
-/**
- * Ozone main class sets up the network servers and initializes the container
- * layer.
- */
-public class OzoneContainer {
-
- public static final Logger LOG = LoggerFactory.getLogger(
- OzoneContainer.class);
-
- private final HddsDispatcher hddsDispatcher;
- private final DatanodeDetails dnDetails;
- private final OzoneConfiguration config;
- private final VolumeSet volumeSet;
- private final ContainerSet containerSet;
- private final Map<ReplicationType, XceiverServerSpi> servers;
-
- /**
- * Construct OzoneContainer object.
- * @param datanodeDetails
- * @param conf
- * @throws DiskOutOfSpaceException
- * @throws IOException
- */
- public OzoneContainer(DatanodeDetails datanodeDetails, OzoneConfiguration
- conf, StateContext context) throws IOException {
- this.dnDetails = datanodeDetails;
- this.config = conf;
- this.volumeSet = new VolumeSet(datanodeDetails.getUuidString(), conf);
- this.containerSet = new ContainerSet();
- buildContainerSet();
- hddsDispatcher = new HddsDispatcher(config, containerSet, volumeSet,
- context);
- servers = new HashMap<>();
- servers.put(ReplicationType.STAND_ALONE,
- new XceiverServerGrpc(datanodeDetails, config, hddsDispatcher,
- createReplicationService()));
- servers.put(ReplicationType.RATIS, XceiverServerRatis
- .newXceiverServerRatis(datanodeDetails, config, hddsDispatcher,
- context));
- }
-
- private GrpcReplicationService createReplicationService() {
- return new GrpcReplicationService(
- new OnDemandContainerReplicationSource(containerSet));
- }
-
- /**
- * Build's container map.
- */
- public void buildContainerSet() {
- Iterator<HddsVolume> volumeSetIterator = volumeSet.getVolumesList()
- .iterator();
- ArrayList<Thread> volumeThreads = new ArrayList<Thread>();
-
- //TODO: diskchecker should be run before this, to see how disks are.
- // And also handle disk failure tolerance need to be added
- while (volumeSetIterator.hasNext()) {
- HddsVolume volume = volumeSetIterator.next();
- File hddsVolumeRootDir = volume.getHddsRootDir();
- Thread thread = new Thread(new ContainerReader(volumeSet, volume,
- containerSet, config));
- thread.start();
- volumeThreads.add(thread);
- }
-
- try {
- for (int i = 0; i < volumeThreads.size(); i++) {
- volumeThreads.get(i).join();
- }
- } catch (InterruptedException ex) {
- LOG.info("Volume Threads Interrupted exception", ex);
- }
-
- }
-
- /**
- * Starts serving requests to ozone container.
- *
- * @throws IOException
- */
- public void start() throws IOException {
- LOG.info("Attempting to start container services.");
- for (XceiverServerSpi serverinstance : servers.values()) {
- serverinstance.start();
- }
- hddsDispatcher.init();
- }
-
- /**
- * Stop Container Service on the datanode.
- */
- public void stop() {
- //TODO: at end of container IO integration work.
- LOG.info("Attempting to stop container services.");
- for(XceiverServerSpi serverinstance: servers.values()) {
- serverinstance.stop();
- }
- hddsDispatcher.shutdown();
- }
-
-
- @VisibleForTesting
- public ContainerSet getContainerSet() {
- return containerSet;
- }
- /**
- * Returns container report.
- * @return - container report.
- * @throws IOException
- */
- public StorageContainerDatanodeProtocolProtos.ContainerReportsProto
- getContainerReport() throws IOException {
- return this.containerSet.getContainerReport();
- }
-
- public PipelineReportsProto getPipelineReport() {
- PipelineReportsProto.Builder pipelineReportsProto =
- PipelineReportsProto.newBuilder();
- for (XceiverServerSpi serverInstance : servers.values()) {
- pipelineReportsProto
- .addAllPipelineReport(serverInstance.getPipelineReport());
- }
- return pipelineReportsProto.build();
- }
-
- /**
- * Submit ContainerRequest.
- * @param request
- * @param replicationType
- * @param pipelineID
- */
- public void submitContainerRequest(
- ContainerProtos.ContainerCommandRequestProto request,
- ReplicationType replicationType,
- PipelineID pipelineID) throws IOException {
- LOG.info("submitting {} request over {} server for container {}",
- request.getCmdType(), replicationType, request.getContainerID());
- Preconditions.checkState(servers.containsKey(replicationType));
- servers.get(replicationType).submitRequest(request, pipelineID);
- }
-
- private int getPortByType(ReplicationType replicationType) {
- return servers.containsKey(replicationType) ?
- servers.get(replicationType).getIPCPort() : INVALID_PORT;
- }
-
- /**
- * Returns the container servers IPC port.
- *
- * @return Container servers IPC port.
- */
- public int getContainerServerPort() {
- return getPortByType(ReplicationType.STAND_ALONE);
- }
-
- /**
- * Returns the Ratis container Server IPC port.
- *
- * @return Ratis port.
- */
- public int getRatisContainerServerPort() {
- return getPortByType(ReplicationType.RATIS);
- }
-
- /**
- * Returns node report of container storage usage.
- */
- public StorageContainerDatanodeProtocolProtos.NodeReportProto getNodeReport()
- throws IOException {
- return volumeSet.getNodeReport();
- }
-
- @VisibleForTesting
- public ContainerDispatcher getDispatcher() {
- return this.hddsDispatcher;
- }
-
- public VolumeSet getVolumeSet() {
- return volumeSet;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/package-info.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/package-info.java
deleted file mode 100644
index c99c038..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/**
- * 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.ozone.container.ozoneimpl;
-/**
- Ozone main that calls into the container layer
-**/
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java
deleted file mode 100644
index 9511241..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- * 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.ozone.container.replication;
-
-import java.io.Closeable;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-
-/**
- * Service to download container data from other datanodes.
- * <p>
- * The implementation of this interface should copy the raw container data in
- * compressed form to working directory.
- * <p>
- * A smart implementation would use multiple sources to do parallel download.
- */
-public interface ContainerDownloader extends Closeable {
-
- CompletableFuture<Path> getContainerDataFromReplicas(long containerId,
- List<DatanodeDetails> sources);
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicationSource.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicationSource.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicationSource.java
deleted file mode 100644
index 69582f7..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicationSource.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.ozone.container.replication;
-
-import java.io.IOException;
-import java.io.OutputStream;
-
-/**
- * Contract to prepare provide the container in binary form..
- * <p>
- * Prepare will be called when container is closed. An implementation could
- * precache any binary representation of a container and store the pre packede
- * images.
- */
-public interface ContainerReplicationSource {
-
- /**
- * Prepare for the replication.
- *
- * @param containerId The name of the container the package.
- */
- void prepare(long containerId);
-
- /**
- * Copy the container data to an output stream.
- *
- * @param containerId Container to replicate
- * @param destination The destination stream to copy all the container data.
- * @throws IOException
- */
- void copyData(long containerId, OutputStream destination)
- throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicator.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicator.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicator.java
deleted file mode 100644
index 827b9d6..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerReplicator.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-/**
- * Service to do the real replication task.
- *
- * An implementation should download the container and im
- */
-public interface ContainerReplicator {
- void replicate(ReplicationTask task);
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerStreamingOutput.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerStreamingOutput.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerStreamingOutput.java
deleted file mode 100644
index f7fd8a4..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerStreamingOutput.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import javax.ws.rs.WebApplicationException;
-import javax.ws.rs.core.StreamingOutput;
-import java.io.IOException;
-import java.io.OutputStream;
-
-/**
- * JAX-RS streaming output to return the binary container data.
- */
-public class ContainerStreamingOutput implements StreamingOutput {
-
- private long containerId;
-
- private ContainerReplicationSource containerReplicationSource;
-
- public ContainerStreamingOutput(long containerId,
- ContainerReplicationSource containerReplicationSource) {
- this.containerId = containerId;
- this.containerReplicationSource = containerReplicationSource;
- }
-
- @Override
- public void write(OutputStream outputStream)
- throws IOException, WebApplicationException {
- containerReplicationSource.copyData(containerId, outputStream);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java
deleted file mode 100644
index 5ef5841..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.io.FileInputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.ozone.container.common.impl.ContainerData;
-import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
-import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
-import org.apache.hadoop.ozone.container.common.interfaces.Container;
-import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
-import org.apache.hadoop.ozone.container.common.interfaces.Handler;
-import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker;
-import org.apache.hadoop.ozone.container.replication.ReplicationTask.Status;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Default replication implementation.
- * <p>
- * This class does the real job. Executes the download and import the container
- * to the container set.
- */
-public class DownloadAndImportReplicator implements ContainerReplicator {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(DownloadAndImportReplicator.class);
-
- private final ContainerSet containerSet;
-
- private final ContainerDispatcher containerDispatcher;
-
- private final ContainerDownloader downloader;
-
- private final TarContainerPacker packer;
-
- public DownloadAndImportReplicator(
- ContainerSet containerSet,
- ContainerDispatcher containerDispatcher,
- ContainerDownloader downloader,
- TarContainerPacker packer) {
- this.containerSet = containerSet;
- this.containerDispatcher = containerDispatcher;
- this.downloader = downloader;
- this.packer = packer;
- }
-
- public void importContainer(long containerID, Path tarFilePath) {
- try {
- ContainerData originalContainerData;
- try (FileInputStream tempContainerTarStream = new FileInputStream(
- tarFilePath.toFile())) {
- byte[] containerDescriptorYaml =
- packer.unpackContainerDescriptor(tempContainerTarStream);
- originalContainerData = ContainerDataYaml.readContainer(
- containerDescriptorYaml);
- }
-
- try (FileInputStream tempContainerTarStream = new FileInputStream(
- tarFilePath.toFile())) {
-
- Handler handler = containerDispatcher.getHandler(
- originalContainerData.getContainerType());
-
- Container container = handler.importContainer(containerID,
- originalContainerData.getMaxSize(),
- tempContainerTarStream,
- packer);
-
- containerSet.addContainer(container);
- }
-
- } catch (Exception e) {
- LOG.error(
- "Can't import the downloaded container data id=" + containerID,
- e);
- try {
- Files.delete(tarFilePath);
- } catch (Exception ex) {
- LOG.error(
- "Container import is failed and the downloaded file can't be "
- + "deleted: "
- + tarFilePath.toAbsolutePath().toString());
- }
- }
- }
-
- @Override
- public void replicate(ReplicationTask task) {
- long containerID = task.getContainerId();
-
- List<DatanodeDetails> sourceDatanodes = task.getSources();
-
- LOG.info("Starting replication of container {} from {}", containerID,
- sourceDatanodes);
-
- CompletableFuture<Path> tempTarFile = downloader
- .getContainerDataFromReplicas(containerID,
- sourceDatanodes);
-
- try {
- //wait for the download. This thread pool is limiting the paralell
- //downloads, so it's ok to block here and wait for the full download.
- Path path = tempTarFile.get();
- LOG.info("Container {} is downloaded, starting to import.",
- containerID);
- importContainer(containerID, path);
- LOG.info("Container {} is replicated successfully", containerID);
- task.setStatus(Status.DONE);
- } catch (Exception e) {
- LOG.error("Container replication was unsuccessful .", e);
- task.setStatus(Status.FAILED);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
deleted file mode 100644
index 3aafb0c..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.io.BufferedOutputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.concurrent.CompletableFuture;
-
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
- .CopyContainerRequestProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
- .CopyContainerResponseProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto
- .IntraDatanodeProtocolServiceGrpc;
-import org.apache.hadoop.hdds.protocol.datanode.proto
- .IntraDatanodeProtocolServiceGrpc.IntraDatanodeProtocolServiceStub;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
-
-import com.google.common.base.Preconditions;
-import org.apache.ratis.shaded.io.grpc.ManagedChannel;
-import org.apache.ratis.shaded.io.grpc.netty.NettyChannelBuilder;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Client to read container data from Grpc.
- */
-public class GrpcReplicationClient {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(GrpcReplicationClient.class);
-
- private final ManagedChannel channel;
-
- private final IntraDatanodeProtocolServiceStub client;
-
- private final Path workingDirectory;
-
- public GrpcReplicationClient(String host,
- int port, Path workingDir) {
-
- channel = NettyChannelBuilder.forAddress(host, port)
- .usePlaintext()
- .maxInboundMessageSize(OzoneConfigKeys.DFS_CONTAINER_CHUNK_MAX_SIZE)
- .build();
- client = IntraDatanodeProtocolServiceGrpc.newStub(channel);
- this.workingDirectory = workingDir;
-
- }
-
- public CompletableFuture<Path> download(long containerId) {
- CopyContainerRequestProto request =
- CopyContainerRequestProto.newBuilder()
- .setContainerID(containerId)
- .setLen(-1)
- .setReadOffset(0)
- .build();
-
- CompletableFuture<Path> response = new CompletableFuture<>();
-
- Path destinationPath =
- getWorkingDirectory().resolve("container-" + containerId + ".tar.gz");
-
- client.download(request,
- new StreamDownloader(containerId, response, destinationPath));
- return response;
- }
-
- private Path getWorkingDirectory() {
- return workingDirectory;
- }
-
- public void shutdown() {
- channel.shutdown();
- }
-
- /**
- * Grpc stream observer to ComletableFuture adapter.
- */
- public static class StreamDownloader
- implements StreamObserver<CopyContainerResponseProto> {
-
- private final CompletableFuture<Path> response;
-
- private final long containerId;
-
- private BufferedOutputStream stream;
-
- private Path outputPath;
-
- public StreamDownloader(long containerId, CompletableFuture<Path> response,
- Path outputPath) {
- this.response = response;
- this.containerId = containerId;
- this.outputPath = outputPath;
- try {
- outputPath = Preconditions.checkNotNull(outputPath);
- Path parentPath = Preconditions.checkNotNull(outputPath.getParent());
- Files.createDirectories(parentPath);
- stream =
- new BufferedOutputStream(new FileOutputStream(outputPath.toFile()));
- } catch (IOException e) {
- throw new RuntimeException("OutputPath can't be used: " + outputPath,
- e);
- }
-
- }
-
- @Override
- public void onNext(CopyContainerResponseProto chunk) {
- try {
- stream.write(chunk.getData().toByteArray());
- } catch (IOException e) {
- response.completeExceptionally(e);
- }
- }
-
- @Override
- public void onError(Throwable throwable) {
- try {
- stream.close();
- LOG.error("Container download was unsuccessfull", throwable);
- try {
- Files.delete(outputPath);
- } catch (IOException ex) {
- LOG.error(
- "Error happened during the download but can't delete the "
- + "temporary destination.", ex);
- }
- response.completeExceptionally(throwable);
- } catch (IOException e) {
- response.completeExceptionally(e);
- }
- }
-
- @Override
- public void onCompleted() {
- try {
- stream.close();
- LOG.info("Container is downloaded to {}", outputPath);
- response.complete(outputPath);
- } catch (IOException e) {
- response.completeExceptionally(e);
- }
-
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
deleted file mode 100644
index d8f696f..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.io.ByteArrayOutputStream;
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
- .CopyContainerRequestProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
- .CopyContainerResponseProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto
- .IntraDatanodeProtocolServiceGrpc;
-
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Service to make containers available for replication.
- */
-public class GrpcReplicationService extends
- IntraDatanodeProtocolServiceGrpc.IntraDatanodeProtocolServiceImplBase {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(GrpcReplicationService.class);
-
- private final ContainerReplicationSource containerReplicationSource;
-
- public GrpcReplicationService(
- ContainerReplicationSource containerReplicationSource) {
- this.containerReplicationSource = containerReplicationSource;
- }
-
- @Override
- public void download(CopyContainerRequestProto request,
- StreamObserver<CopyContainerResponseProto> responseObserver) {
- LOG.info("Streaming container data ({}) to other datanode",
- request.getContainerID());
- try {
- GrpcOutputStream outputStream =
- new GrpcOutputStream(responseObserver, request.getContainerID());
- containerReplicationSource
- .copyData(request.getContainerID(), outputStream);
-
- } catch (IOException e) {
- LOG.error("Can't stream the container data", e);
- responseObserver.onError(e);
- }
- }
-
- private static class GrpcOutputStream extends OutputStream
- implements Closeable {
-
- private static final int BUFFER_SIZE_IN_BYTES = 1024 * 1024;
-
- private final StreamObserver<CopyContainerResponseProto> responseObserver;
-
- private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-
- private long containerId;
-
- private int readOffset = 0;
-
- private int writtenBytes;
-
- GrpcOutputStream(
- StreamObserver<CopyContainerResponseProto> responseObserver,
- long containerId) {
- this.responseObserver = responseObserver;
- this.containerId = containerId;
- }
-
- @Override
- public void write(int b) throws IOException {
- try {
- buffer.write(b);
- if (buffer.size() > BUFFER_SIZE_IN_BYTES) {
- flushBuffer(false);
- }
- } catch (Exception ex) {
- responseObserver.onError(ex);
- }
- }
-
- private void flushBuffer(boolean eof) {
- if (buffer.size() > 0) {
- CopyContainerResponseProto response =
- CopyContainerResponseProto.newBuilder()
- .setContainerID(containerId)
- .setData(ByteString.copyFrom(buffer.toByteArray()))
- .setEof(eof)
- .setReadOffset(readOffset)
- .setLen(buffer.size())
- .build();
- responseObserver.onNext(response);
- readOffset += buffer.size();
- writtenBytes += buffer.size();
- buffer.reset();
- }
- }
-
- @Override
- public void close() throws IOException {
- flushBuffer(true);
- LOG.info("{} bytes written to the rpc stream from container {}",
- writtenBytes, containerId);
- responseObserver.onCompleted();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java
deleted file mode 100644
index d557b54..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
-import org.apache.hadoop.ozone.container.common.interfaces.Container;
-import org.apache.hadoop.ozone.container.common.interfaces.ContainerPacker;
-import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker;
-
-import com.google.common.base.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A naive implementation of the replication source which creates a tar file
- * on-demand without pre-create the compressed archives.
- */
-public class OnDemandContainerReplicationSource
- implements ContainerReplicationSource {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(ContainerReplicationSource.class);
-
- private ContainerSet containerSet;
-
- private ContainerPacker packer = new TarContainerPacker();
-
- public OnDemandContainerReplicationSource(
- ContainerSet containerSet) {
- this.containerSet = containerSet;
- }
-
- @Override
- public void prepare(long containerId) {
-
- }
-
- @Override
- public void copyData(long containerId, OutputStream destination)
- throws IOException {
-
- Container container = containerSet.getContainer(containerId);
-
- Preconditions
- .checkNotNull(container, "Container is not found " + containerId);
-
- switch (container.getContainerType()) {
- case KeyValueContainer:
- packer.pack(container,
- destination);
- break;
- default:
- LOG.warn("Container type " + container.getContainerType()
- + " is not replicable as no compression algorithm for that.");
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
deleted file mode 100644
index 1d8d5f6..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-
-import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
-import org.apache.hadoop.ozone.container.replication.ReplicationTask.Status;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Single point to schedule the downloading tasks based on priorities.
- */
-public class ReplicationSupervisor {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(ReplicationSupervisor.class);
-
- private final Set<Worker> threadPool = new HashSet<>();
-
- private final Map<Long, ReplicationTask> queue = new TreeMap();
-
- private final ContainerSet containerSet;
-
- private final ContainerReplicator replicator;
-
- private final int poolSize;
-
- public ReplicationSupervisor(
- ContainerSet containerSet,
- ContainerReplicator replicator, int poolSize) {
- this.containerSet = containerSet;
- this.replicator = replicator;
- this.poolSize = poolSize;
- }
-
- public synchronized void addTask(ReplicationTask task) {
- queue.putIfAbsent(task.getContainerId(), task);
- synchronized (threadPool) {
- threadPool.notify();
- }
- }
-
- public void start() {
- for (int i = 0; i < poolSize; i++) {
- Worker worker = new Worker();
- Thread thread = new Thread(worker, "ContainerReplication-" + i);
- thread.setDaemon(true);
- thread.start();
- threadPool.add(worker);
- }
- }
-
- public synchronized ReplicationTask selectTask() {
- for (ReplicationTask task : queue.values()) {
- if (task.getStatus() == Status.QUEUED) {
- if (containerSet.getContainer(task.getContainerId()) == null) {
- task.setStatus(Status.DOWNLOADING);
- return task;
- } else {
- LOG.debug("Container {} has already been downloaded.",
- task.getContainerId());
- queue.remove(task.getContainerId());
- }
- } else if (task.getStatus() == Status.FAILED) {
- LOG.error(
- "Container {} can't be downloaded from any of the datanodes.",
- task.getContainerId());
- queue.remove(task.getContainerId());
- } else if (task.getStatus() == Status.DONE) {
- queue.remove(task.getContainerId());
- LOG.info("Container {} is replicated.", task.getContainerId());
- }
- }
- //no available task.
- return null;
- }
-
- public void stop() {
- for (Worker worker : threadPool) {
- worker.stop();
- }
- }
-
- @VisibleForTesting
- public int getQueueSize() {
- return queue.size();
- }
-
- private class Worker implements Runnable {
-
- private boolean running = true;
-
- @Override
- public void run() {
- try {
- while (running) {
- ReplicationTask task = selectTask();
- if (task == null) {
- synchronized (threadPool) {
- threadPool.wait();
- }
- } else {
- replicator.replicate(task);
- }
- }
- } catch (Exception ex) {
- LOG.error("Error on doing replication", ex);
- try {
- Thread.sleep(200);
- } catch (InterruptedException e) {
- LOG.error("Error on waiting after failed replication task", e);
- }
- }
- }
-
- public void stop() {
- running = false;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java
deleted file mode 100644
index 9019811..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.container.replication;
-
-import java.time.Instant;
-import java.util.List;
-import java.util.Objects;
-
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-
-/**
- * The task to download a container from the sources.
- */
-public class ReplicationTask {
-
- private volatile Status status = Status.QUEUED;
-
- private final long containerId;
-
- private List<DatanodeDetails> sources;
-
- private final Instant queued = Instant.now();
-
- public ReplicationTask(long containerId,
- List<DatanodeDetails> sources) {
- this.containerId = containerId;
- this.sources = sources;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ReplicationTask that = (ReplicationTask) o;
- return containerId == that.containerId;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(containerId);
- }
-
- public long getContainerId() {
- return containerId;
- }
-
- public List<DatanodeDetails> getSources() {
- return sources;
- }
-
- public Status getStatus() {
- return status;
- }
-
- public void setStatus(
- Status status) {
- this.status = status;
- }
-
- @Override
- public String toString() {
- return "ReplicationTask{" +
- "status=" + status +
- ", containerId=" + containerId +
- ", sources=" + sources +
- ", queued=" + queued +
- '}';
- }
-
- public Instant getQueued() {
- return queued;
- }
-
- /**
- * Status of the replication.
- */
- public enum Status {
- QUEUED,
- DOWNLOADING,
- FAILED,
- DONE
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java
deleted file mode 100644
index a461a98..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * 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.ozone.container.replication;
-
-import java.io.IOException;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Simple ContainerDownloaderImplementation to download the missing container
- * from the first available datanode.
- * <p>
- * This is not the most effective implementation as it uses only one source
- * for he container download.
- */
-public class SimpleContainerDownloader implements ContainerDownloader {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(SimpleContainerDownloader.class);
-
- private final Path workingDirectory;
-
- private ExecutorService executor;
-
- public SimpleContainerDownloader(Configuration conf) {
-
- String workDirString =
- conf.get(OzoneConfigKeys.OZONE_CONTAINER_COPY_WORKDIR);
-
- if (workDirString == null) {
- workingDirectory = Paths.get(System.getProperty("java.io.tmpdir"))
- .resolve("container-copy");
- } else {
- workingDirectory = Paths.get(workDirString);
- }
-
- ThreadFactory build = new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("Container downloader thread - %d").build();
- executor = Executors.newSingleThreadExecutor(build);
- LOG.info("Starting container downloader service to copy "
- + "containers to replicate.");
- }
-
- @Override
- public CompletableFuture<Path> getContainerDataFromReplicas(long containerId,
- List<DatanodeDetails> sourceDatanodes) {
-
- CompletableFuture<Path> result = null;
- for (DatanodeDetails datanode : sourceDatanodes) {
- try {
-
- if (result == null) {
- GrpcReplicationClient grpcReplicationClient =
- new GrpcReplicationClient(datanode.getIpAddress(),
- datanode.getPort(Name.STANDALONE).getValue(),
- workingDirectory);
- result = grpcReplicationClient.download(containerId);
- } else {
- result = result.thenApply(CompletableFuture::completedFuture)
- .exceptionally(t -> {
- LOG.error("Error on replicating container: " + containerId, t);
- GrpcReplicationClient grpcReplicationClient =
- new GrpcReplicationClient(datanode.getIpAddress(),
- datanode.getPort(Name.STANDALONE).getValue(),
- workingDirectory);
- return grpcReplicationClient.download(containerId);
- }).thenCompose(Function.identity());
- }
- } catch (Exception ex) {
- LOG.error(String.format(
- "Container %s download from datanode %s was unsuccessful. "
- + "Trying the next datanode", containerId, datanode), ex);
- }
-
- }
- return result;
-
- }
-
- @Override
- public void close() throws IOException {
- try {
- executor.awaitTermination(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- LOG.error("Can't stop container downloader gracefully", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/package-info.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/package-info.java
deleted file mode 100644
index 38a853c..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/**
- * 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.ozone.container.replication;
-/**
- Classes to replicate container data between datanodes.
-**/
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/package-info.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/package-info.java
deleted file mode 100644
index 1a51012..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/package-info.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/**
- * 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.ozone;
-
-/**
- * Generic ozone specific classes.
- */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java
deleted file mode 100644
index e3b5370..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * 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.ozone.protocol;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.NodeReportProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMVersionRequestProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMVersionResponseProto;
-
-import java.io.IOException;
-
-/**
- * The protocol spoken between datanodes and SCM. For specifics please the
- * Protoc file that defines this protocol.
- */
-@InterfaceAudience.Private
-public interface StorageContainerDatanodeProtocol {
- /**
- * Returns SCM version.
- * @return Version info.
- */
- SCMVersionResponseProto getVersion(SCMVersionRequestProto versionRequest)
- throws IOException;
-
- /**
- * Used by data node to send a Heartbeat.
- * @param heartbeat Heartbeat
- * @return - SCMHeartbeatResponseProto
- * @throws IOException
- */
- SCMHeartbeatResponseProto sendHeartbeat(SCMHeartbeatRequestProto heartbeat)
- throws IOException;
-
- /**
- * Register Datanode.
- * @param datanodeDetails - Datanode Details.
- * @param nodeReport - Node Report.
- * @param containerReportsRequestProto - Container Reports.
- * @return SCM Command.
- */
- SCMRegisteredResponseProto register(
- DatanodeDetailsProto datanodeDetails,
- NodeReportProto nodeReport,
- ContainerReportsProto containerReportsRequestProto,
- PipelineReportsProto pipelineReports) throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerNodeProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerNodeProtocol.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerNodeProtocol.java
deleted file mode 100644
index b3c3eb3..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerNodeProtocol.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * 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.ozone.protocol;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.NodeReportProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMVersionRequestProto;
-import org.apache.hadoop.ozone.protocol.commands.RegisteredCommand;
-import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
-
-import java.util.List;
-
-/**
- * The protocol spoken between datanodes and SCM.
- *
- * Please note that the full protocol spoken between a datanode and SCM is
- * separated into 2 interfaces. One interface that deals with node state and
- * another interface that deals with containers.
- *
- * This interface has functions that deals with the state of datanode.
- */
-@InterfaceAudience.Private
-public interface StorageContainerNodeProtocol {
- /**
- * Gets the version info from SCM.
- * @param versionRequest - version Request.
- * @return - returns SCM version info and other required information needed
- * by datanode.
- */
- VersionResponse getVersion(SCMVersionRequestProto versionRequest);
-
- /**
- * Register the node if the node finds that it is not registered with any SCM.
- * @param datanodeDetails DatanodeDetails
- * @param nodeReport NodeReportProto
- * @param pipelineReport PipelineReportsProto
- * @return SCMHeartbeatResponseProto
- */
- RegisteredCommand register(DatanodeDetails datanodeDetails,
- NodeReportProto nodeReport,
- PipelineReportsProto pipelineReport);
-
- /**
- * Send heartbeat to indicate the datanode is alive and doing well.
- * @param datanodeDetails - Datanode ID.
- * @return SCMheartbeat response list
- */
- List<SCMCommand> processHeartbeat(DatanodeDetails datanodeDetails);
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/VersionResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/VersionResponse.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/VersionResponse.java
deleted file mode 100644
index 4d328d3..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/VersionResponse.java
+++ /dev/null
@@ -1,154 +0,0 @@
-
-/**
- * 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.ozone.protocol;
-
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.KeyValue;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMVersionResponseProto;
-
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-/**
- * Version response class.
- */
-public class VersionResponse {
- private final int version;
- private final Map<String, String> values;
-
- /**
- * Creates a version response class.
- * @param version
- * @param values
- */
- public VersionResponse(int version, Map<String, String> values) {
- this.version = version;
- this.values = values;
- }
-
- /**
- * Creates a version Response class.
- * @param version
- */
- public VersionResponse(int version) {
- this.version = version;
- this.values = new HashMap<>();
- }
-
- /**
- * Returns a new Builder.
- * @return - Builder.
- */
- public static Builder newBuilder() {
- return new Builder();
- }
-
- /**
- * Returns this class from protobuf message.
- * @param response - SCMVersionResponseProto
- * @return VersionResponse
- */
- public static VersionResponse getFromProtobuf(SCMVersionResponseProto
- response) {
- return new VersionResponse(response.getSoftwareVersion(),
- response.getKeysList().stream()
- .collect(Collectors.toMap(KeyValue::getKey,
- KeyValue::getValue)));
- }
-
- /**
- * Adds a value to version Response.
- * @param key - String
- * @param value - String
- */
- public void put(String key, String value) {
- if (this.values.containsKey(key)) {
- throw new IllegalArgumentException("Duplicate key in version response");
- }
- values.put(key, value);
- }
-
- /**
- * Return a protobuf message.
- * @return SCMVersionResponseProto.
- */
- public SCMVersionResponseProto getProtobufMessage() {
-
- List<KeyValue> list = new LinkedList<>();
- for (Map.Entry<String, String> entry : values.entrySet()) {
- list.add(KeyValue.newBuilder().setKey(entry.getKey()).
- setValue(entry.getValue()).build());
- }
- return
- SCMVersionResponseProto.newBuilder()
- .setSoftwareVersion(this.version)
- .addAllKeys(list).build();
- }
-
- public String getValue(String key) {
- return this.values.get(key);
- }
-
- /**
- * Builder class.
- */
- public static class Builder {
- private int version;
- private Map<String, String> values;
-
- Builder() {
- values = new HashMap<>();
- }
-
- /**
- * Sets the version.
- * @param ver - version
- * @return Builder
- */
- public Builder setVersion(int ver) {
- this.version = ver;
- return this;
- }
-
- /**
- * Adds a value to version Response.
- * @param key - String
- * @param value - String
- */
- public Builder addValue(String key, String value) {
- if (this.values.containsKey(key)) {
- throw new IllegalArgumentException("Duplicate key in version response");
- }
- values.put(key, value);
- return this;
- }
-
- /**
- * Builds the version response.
- * @return VersionResponse.
- */
- public VersionResponse build() {
- return new VersionResponse(this.version, this.values);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CloseContainerCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CloseContainerCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CloseContainerCommand.java
deleted file mode 100644
index c2c20a4..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CloseContainerCommand.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.protocol.commands;
-
-import com.google.common.base.Preconditions;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMCommandProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.CloseContainerCommandProto;
-import org.apache.hadoop.hdds.scm.container.common.helpers.PipelineID;
-
-/**
- * Asks datanode to close a container.
- */
-public class CloseContainerCommand
- extends SCMCommand<CloseContainerCommandProto> {
-
- private HddsProtos.ReplicationType replicationType;
- private PipelineID pipelineID;
-
- public CloseContainerCommand(long containerID,
- HddsProtos.ReplicationType replicationType,
- PipelineID pipelineID) {
- super(containerID);
- this.replicationType = replicationType;
- this.pipelineID = pipelineID;
- }
-
- /**
- * Returns the type of this command.
- *
- * @return Type
- */
- @Override
- public SCMCommandProto.Type getType() {
- return SCMCommandProto.Type.closeContainerCommand;
- }
-
- /**
- * Gets the protobuf message of this object.
- *
- * @return A protobuf message.
- */
- @Override
- public byte[] getProtoBufMessage() {
- return getProto().toByteArray();
- }
-
- public CloseContainerCommandProto getProto() {
- return CloseContainerCommandProto.newBuilder()
- .setContainerID(getId())
- .setCmdId(getId())
- .setReplicationType(replicationType)
- .setPipelineID(pipelineID.getProtobuf())
- .build();
- }
-
- public static CloseContainerCommand getFromProtobuf(
- CloseContainerCommandProto closeContainerProto) {
- Preconditions.checkNotNull(closeContainerProto);
- return new CloseContainerCommand(closeContainerProto.getCmdId(),
- closeContainerProto.getReplicationType(),
- PipelineID.getFromProtobuf(closeContainerProto.getPipelineID()));
- }
-
- public long getContainerID() {
- return getId();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandForDatanode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandForDatanode.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandForDatanode.java
deleted file mode 100644
index 69337fb..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandForDatanode.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.protocol.commands;
-
-import java.util.UUID;
-
-import com.google.protobuf.GeneratedMessage;
-import org.apache.hadoop.hdds.server.events.IdentifiableEventPayload;
-
-/**
- * Command for the datanode with the destination address.
- */
-public class CommandForDatanode<T extends GeneratedMessage> implements
- IdentifiableEventPayload {
-
- private final UUID datanodeId;
-
- private final SCMCommand<T> command;
-
- public CommandForDatanode(UUID datanodeId, SCMCommand<T> command) {
- this.datanodeId = datanodeId;
- this.command = command;
- }
-
- public UUID getDatanodeId() {
- return datanodeId;
- }
-
- public SCMCommand<T> getCommand() {
- return command;
- }
-
- public long getId() {
- return command.getId();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandStatus.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandStatus.java
deleted file mode 100644
index 4b3ce84..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/CommandStatus.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.protocol.commands;
-
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.CommandStatus.Status;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMCommandProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type;
-
-/**
- * A class that is used to communicate status of datanode commands.
- */
-public class CommandStatus {
-
- private SCMCommandProto.Type type;
- private Long cmdId;
- private Status status;
- private String msg;
-
- CommandStatus(Type type, Long cmdId, Status status, String msg) {
- this.type = type;
- this.cmdId = cmdId;
- this.status = status;
- this.msg = msg;
- }
-
- public Type getType() {
- return type;
- }
-
- public Long getCmdId() {
- return cmdId;
- }
-
- public Status getStatus() {
- return status;
- }
-
- public String getMsg() {
- return msg;
- }
-
- /**
- * To allow change of status once commandStatus is initialized.
- *
- * @param status
- */
- public void setStatus(Status status) {
- this.status = status;
- }
-
- public void setStatus(boolean cmdExecuted) {
- setStatus(cmdExecuted ? Status.EXECUTED : Status.FAILED);
- }
-
- /**
- * Returns a CommandStatus from the protocol buffers.
- *
- * @param cmdStatusProto - protoBuf Message
- * @return CommandStatus
- */
- public CommandStatus getFromProtoBuf(
- StorageContainerDatanodeProtocolProtos.CommandStatus cmdStatusProto) {
- return CommandStatusBuilder.newBuilder()
- .setCmdId(cmdStatusProto.getCmdId())
- .setStatus(cmdStatusProto.getStatus())
- .setType(cmdStatusProto.getType())
- .setMsg(cmdStatusProto.getMsg())
- .build();
- }
- /**
- * Returns a CommandStatus from the protocol buffers.
- *
- * @return StorageContainerDatanodeProtocolProtos.CommandStatus
- */
- public StorageContainerDatanodeProtocolProtos.CommandStatus
- getProtoBufMessage() {
- StorageContainerDatanodeProtocolProtos.CommandStatus.Builder builder =
- StorageContainerDatanodeProtocolProtos.CommandStatus.newBuilder()
- .setCmdId(this.getCmdId())
- .setStatus(this.getStatus())
- .setType(this.getType());
- if (this.getMsg() != null) {
- builder.setMsg(this.getMsg());
- }
- return builder.build();
- }
-
- /**
- * Builder class for CommandStatus.
- */
- public static class CommandStatusBuilder {
-
- private SCMCommandProto.Type type;
- private Long cmdId;
- private StorageContainerDatanodeProtocolProtos.CommandStatus.Status status;
- private String msg;
-
- CommandStatusBuilder() {
- }
-
- public static CommandStatusBuilder newBuilder() {
- return new CommandStatusBuilder();
- }
-
- public Type getType() {
- return type;
- }
-
- public Long getCmdId() {
- return cmdId;
- }
-
- public Status getStatus() {
- return status;
- }
-
- public String getMsg() {
- return msg;
- }
-
- public CommandStatusBuilder setType(Type commandType) {
- this.type = commandType;
- return this;
- }
-
- public CommandStatusBuilder setCmdId(Long commandId) {
- this.cmdId = commandId;
- return this;
- }
-
- public CommandStatusBuilder setStatus(Status commandStatus) {
- this.status = commandStatus;
- return this;
- }
-
- public CommandStatusBuilder setMsg(String message) {
- this.msg = message;
- return this;
- }
-
- public CommandStatus build() {
- return new CommandStatus(type, cmdId, status, msg);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlockCommandStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlockCommandStatus.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlockCommandStatus.java
deleted file mode 100644
index 2659ab3..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlockCommandStatus.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.ozone.protocol.commands;
-
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto;
-
-public class DeleteBlockCommandStatus extends CommandStatus {
-
- private ContainerBlocksDeletionACKProto blocksDeletionAck = null;
-
- public DeleteBlockCommandStatus(Type type, Long cmdId,
- StorageContainerDatanodeProtocolProtos.CommandStatus.Status status,
- String msg, ContainerBlocksDeletionACKProto blocksDeletionAck) {
- super(type, cmdId, status, msg);
- this.blocksDeletionAck = blocksDeletionAck;
- }
-
- public void setBlocksDeletionAck(
- ContainerBlocksDeletionACKProto deletionAck) {
- blocksDeletionAck = deletionAck;
- }
-
- @Override
- public CommandStatus getFromProtoBuf(
- StorageContainerDatanodeProtocolProtos.CommandStatus cmdStatusProto) {
- return DeleteBlockCommandStatusBuilder.newBuilder()
- .setBlockDeletionAck(cmdStatusProto.getBlockDeletionAck())
- .setCmdId(cmdStatusProto.getCmdId())
- .setStatus(cmdStatusProto.getStatus())
- .setType(cmdStatusProto.getType())
- .setMsg(cmdStatusProto.getMsg())
- .build();
- }
-
- @Override
- public StorageContainerDatanodeProtocolProtos.CommandStatus getProtoBufMessage() {
- StorageContainerDatanodeProtocolProtos.CommandStatus.Builder builder =
- StorageContainerDatanodeProtocolProtos.CommandStatus.newBuilder()
- .setCmdId(this.getCmdId())
- .setStatus(this.getStatus())
- .setType(this.getType());
- if (blocksDeletionAck != null) {
- builder.setBlockDeletionAck(blocksDeletionAck);
- }
- if (this.getMsg() != null) {
- builder.setMsg(this.getMsg());
- }
- return builder.build();
- }
-
- public static final class DeleteBlockCommandStatusBuilder
- extends CommandStatusBuilder {
- private ContainerBlocksDeletionACKProto blocksDeletionAck = null;
-
- public static DeleteBlockCommandStatusBuilder newBuilder() {
- return new DeleteBlockCommandStatusBuilder();
- }
-
- public DeleteBlockCommandStatusBuilder setBlockDeletionAck(
- ContainerBlocksDeletionACKProto deletionAck) {
- this.blocksDeletionAck = deletionAck;
- return this;
- }
-
- @Override
- public CommandStatus build() {
- return new DeleteBlockCommandStatus(getType(), getCmdId(), getStatus(),
- getMsg(), blocksDeletionAck);
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2c392da8/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlocksCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlocksCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlocksCommand.java
deleted file mode 100644
index 07feeff..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/DeleteBlocksCommand.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * 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.ozone.protocol.commands;
-
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.SCMCommandProto;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
-import org.apache.hadoop.hdds.protocol.proto
- .StorageContainerDatanodeProtocolProtos.DeleteBlocksCommandProto;
-
-import java.util.List;
-
-/**
- * A SCM command asks a datanode to delete a number of blocks.
- */
-public class DeleteBlocksCommand extends
- SCMCommand<DeleteBlocksCommandProto> {
-
- private List<DeletedBlocksTransaction> blocksTobeDeleted;
-
-
- public DeleteBlocksCommand(List<DeletedBlocksTransaction> blocks) {
- super();
- this.blocksTobeDeleted = blocks;
- }
-
- // Should be called only for protobuf conversion
- private DeleteBlocksCommand(List<DeletedBlocksTransaction> blocks,
- long id) {
- super(id);
- this.blocksTobeDeleted = blocks;
- }
-
- public List<DeletedBlocksTransaction> blocksTobeDeleted() {
- return this.blocksTobeDeleted;
- }
-
- @Override
- public SCMCommandProto.Type getType() {
- return SCMCommandProto.Type.deleteBlocksCommand;
- }
-
- @Override
- public byte[] getProtoBufMessage() {
- return getProto().toByteArray();
- }
-
- public static DeleteBlocksCommand getFromProtobuf(
- DeleteBlocksCommandProto deleteBlocksProto) {
- return new DeleteBlocksCommand(deleteBlocksProto
- .getDeletedBlocksTransactionsList(), deleteBlocksProto.getCmdId());
- }
-
- public DeleteBlocksCommandProto getProto() {
- return DeleteBlocksCommandProto.newBuilder()
- .setCmdId(getId())
- .addAllDeletedBlocksTransactions(blocksTobeDeleted).build();
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org