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 ae...@apache.org on 2018/03/15 16:32:29 UTC
[14/51] [partial] hadoop git commit: HDFS-13258. Ozone: restructure
Hdsl/Ozone code to separated maven subprojects. Contributed by Elek Marton,
Mukul Kumar Singh, Xiaoyu Yao, Ajay Kumar, Anu Engineer, Lokesh Jain,
Nanda Kumar.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/OzoneMetadataManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/OzoneMetadataManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/OzoneMetadataManager.java
deleted file mode 100644
index 1fe9a18..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/OzoneMetadataManager.java
+++ /dev/null
@@ -1,1138 +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.web.localstorage;
-
-import com.google.common.base.Preconditions;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.client.io.LengthInputStream;
-import org.apache.hadoop.ozone.web.exceptions.ErrorTable;
-import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.hadoop.ozone.web.handlers.BucketArgs;
-import org.apache.hadoop.ozone.web.handlers.KeyArgs;
-import org.apache.hadoop.ozone.web.handlers.ListArgs;
-import org.apache.hadoop.ozone.web.handlers.UserArgs;
-import org.apache.hadoop.ozone.web.handlers.VolumeArgs;
-import org.apache.hadoop.ozone.OzoneAcl;
-import org.apache.hadoop.ozone.web.response.BucketInfo;
-import org.apache.hadoop.ozone.web.response.KeyInfo;
-import org.apache.hadoop.ozone.web.response.ListBuckets;
-import org.apache.hadoop.ozone.web.response.ListKeys;
-import org.apache.hadoop.ozone.web.response.ListVolumes;
-import org.apache.hadoop.ozone.web.response.VolumeInfo;
-import org.apache.hadoop.ozone.web.response.VolumeOwner;
-import org.apache.hadoop.utils.MetadataStore;
-import org.apache.hadoop.utils.MetadataStoreBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.charset.Charset;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Locale;
-import java.util.TimeZone;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-/**
- * A stand alone Ozone implementation that allows us to run Ozone tests in local
- * mode. This acts as the ozone backend when using MiniDFSCluster for testing.
- */
-public final class OzoneMetadataManager {
-
- /*
- OzoneMetadataManager manages volume/bucket/object metadata and
- data.
-
- Metadata is maintained in 2 level DB files, UserDB and MetadataDB.
-
- UserDB contains a Name and a List. For example volumes owned by the user
- bilbo, would be maintained in UserDB as {bilbo}->{shire, rings}
-
- This list part of mapping is context sensitive. That is, if you use {user
- name} as the key, the list you get is a list of volumes. if you use
- {user/volume} as the key the list you get is list of buckets. if you use
- {user/volume/bucket} as key the list you get is the list of objects.
-
- All keys in the UserDB starts with the UserName.
-
- We also need to maintain a flat namespace for volumes. This is
- maintained by the MetadataDB. MetadataDB contains the name of an
- object(volume, bucket or key) and its associated metadata.
- The keys in the Metadata DB are {volume}, {volume/bucket} or
- {volume/bucket/key}. User name is absent, so we have a common root name
- space for the volume.
-
- The value of part of metadataDB points to corresponding *Info structures.
- {volume] -> volumeInfo
- {volume/bucket} -> bucketInfo
- {volume/bucket/key} -> keyInfo
-
-
- Here are various work flows :
-
- CreateVolume -> Check if Volume exists in metadataDB, if not update UserDB
- with a list of volumes and update metadataDB with VolumeInfo.
-
- DeleteVolume -> Check the Volume, and check the VolumeInfo->bucketCount.
- if bucketCount == 0, delete volume from userDB->{List of volumes} and
- metadataDB.
-
- Very similar work flows exist for CreateBucket and DeleteBucket.
-
- // Please note : These database operations are *not* transactional,
- // which means that failure can lead to inconsistencies.
- // Only way to recover is to reset to a clean state, or
- // use rm -rf /tmp/ozone :)
-
- We have very simple locking policy. We have a ReaderWriter lock that is
- taken for each action, this lock is aptly named "lock".
-
- All actions *must* be performed with a lock held, either a read
- lock or a write lock. Violation of these locking policies can be harmful.
-
-
- // // IMPORTANT :
- // // This is a simulation layer, this is NOT how the real
- // // OZONE functions. This is written to so that we can write
- // // stand-alone tests for the protocol and client code.
-
-*/
- static final Logger LOG = LoggerFactory.getLogger(OzoneMetadataManager.class);
- private static final String USER_DB = "/user.db";
- private static final String META_DB = "/metadata.db";
- private static OzoneMetadataManager bm = null;
- private MetadataStore userDB;
- private MetadataStore metadataDB;
- private ReadWriteLock lock;
- private Charset encoding = Charset.forName("UTF-8");
- private String storageRoot;
- private static final String OBJECT_DIR = "/_objects/";
-
- // This table keeps a pointer to objects whose operations
- // are in progress but not yet committed to persistent store
- private ConcurrentHashMap<OutputStream, String> inProgressObjects;
-
- /**
- * Constructs OzoneMetadataManager.
- */
- private OzoneMetadataManager(Configuration conf) throws IOException {
-
- lock = new ReentrantReadWriteLock();
- storageRoot =
- conf.getTrimmed(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT,
- OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT);
-
- File file = new File(storageRoot + OBJECT_DIR);
-
- if (!file.exists() && !file.mkdirs()) {
- LOG.error("Creation of Ozone root failed. " + file.toString());
- throw new IOException("Creation of Ozone root failed.");
- }
-
- try {
- userDB = MetadataStoreBuilder.newBuilder()
- .setDbFile(new File(storageRoot + USER_DB))
- .setCreateIfMissing(true)
- .build();
- metadataDB = MetadataStoreBuilder.newBuilder()
- .setDbFile(new File(storageRoot + META_DB))
- .setCreateIfMissing(true)
- .build();
- inProgressObjects = new ConcurrentHashMap<>();
- } catch (IOException ex) {
- LOG.error("Cannot open db :" + ex.getMessage());
- throw ex;
- }
- }
-
- /**
- * Gets Ozone Manager.
- *
- * @return OzoneMetadataManager
- */
- public static synchronized OzoneMetadataManager
- getOzoneMetadataManager(Configuration conf) throws IOException {
- if (bm == null) {
- bm = new OzoneMetadataManager(conf);
- }
- return bm;
- }
-
- /**
- * Creates a volume.
- *
- * @param args - VolumeArgs
- * @throws OzoneException
- */
- public void createVolume(VolumeArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- SimpleDateFormat format =
- new SimpleDateFormat(OzoneConsts.OZONE_DATE_FORMAT, Locale.US);
- format.setTimeZone(TimeZone.getTimeZone(OzoneConsts.OZONE_TIME_ZONE));
-
- byte[] volumeName =
- metadataDB.get(args.getVolumeName().getBytes(encoding));
-
- if (volumeName != null) {
- LOG.debug("Volume {} already exists.", volumeName);
- throw ErrorTable.newError(ErrorTable.VOLUME_ALREADY_EXISTS, args);
- }
-
- VolumeInfo newVInfo = new VolumeInfo(args.getVolumeName(), format
- .format(new Date(System.currentTimeMillis())), args.getAdminName());
-
- newVInfo.setQuota(args.getQuota());
- VolumeOwner owner = new VolumeOwner(args.getUserName());
- newVInfo.setOwner(owner);
-
- ListVolumes volumeList;
- byte[] userVolumes = userDB.get(args.getUserName().getBytes(encoding));
- if (userVolumes == null) {
- volumeList = new ListVolumes();
- } else {
- volumeList = ListVolumes.parse(new String(userVolumes, encoding));
- }
-
- volumeList.addVolume(newVInfo);
- volumeList.sort();
-
- // Please note : These database operations are *not* transactional,
- // which means that failure can lead to inconsistencies.
- // Only way to recover is to reset to a clean state, or
- // use rm -rf /tmp/ozone :)
-
-
- userDB.put(args.getUserName().getBytes(encoding),
- volumeList.toDBString().getBytes(encoding));
-
- metadataDB.put(args.getVolumeName().getBytes(encoding),
- newVInfo.toDBString().getBytes(encoding));
-
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Updates the Volume properties like Owner Name and Quota.
- *
- * @param args - Volume Args
- * @param property - Flag which tells us what property to upgrade
- * @throws OzoneException
- */
- public void setVolumeProperty(VolumeArgs args, VolumeProperty property)
- throws OzoneException {
- lock.writeLock().lock();
- try {
- byte[] volumeInfo =
- metadataDB.get(args.getVolumeName().getBytes(encoding));
- if (volumeInfo == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, args);
- }
- VolumeInfo info = VolumeInfo.parse(new String(volumeInfo, encoding));
-
- byte[] userBytes = userDB.get(args.getResourceName().getBytes(encoding));
- ListVolumes volumeList;
- if (userBytes == null) {
- volumeList = new ListVolumes();
- } else {
- volumeList = ListVolumes.parse(new String(userBytes, encoding));
- }
-
- switch (property) {
- case OWNER:
- // needs new owner, we delete the volume object from the
- // old user's volume list
- removeOldOwner(info);
- VolumeOwner owner = new VolumeOwner(args.getUserName());
- // set the new owner
- info.setOwner(owner);
- break;
- case QUOTA:
- // if this is quota update we just remove the old object from the
- // current users list and update the same object later.
- volumeList.getVolumes().remove(info);
- info.setQuota(args.getQuota());
- break;
- default:
- OzoneException ozEx =
- ErrorTable.newError(ErrorTable.BAD_PROPERTY, args);
- ozEx.setMessage("Volume property is not recognized");
- throw ozEx;
- }
-
- volumeList.addVolume(info);
-
- metadataDB.put(args.getVolumeName().getBytes(encoding),
- info.toDBString().getBytes(encoding));
-
- // if this is an owner change this put will create a new owner or update
- // the owner's volume list.
- userDB.put(args.getResourceName().getBytes(encoding),
- volumeList.toDBString().getBytes(encoding));
-
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Removes the old owner from the volume.
- *
- * @param info - VolumeInfo
- * @throws IOException
- */
- private void removeOldOwner(VolumeInfo info) throws IOException {
- // We need to look the owner that we know is the current owner
- byte[] volumeBytes =
- userDB.get(info.getOwner().getName().getBytes(encoding));
- ListVolumes volumeList =
- ListVolumes.parse(new String(volumeBytes, encoding));
- volumeList.getVolumes().remove(info);
-
- // Write the new list info to the old user data
- userDB.put(info.getOwner().getName().getBytes(encoding),
- volumeList.toDBString().getBytes(encoding));
- }
-
- /**
- * Checks if you are the owner of a specific volume.
- *
- * @param volume - Volume Name whose access permissions needs to be checked
- * @param acl - requested acls which needs to be checked for access
- * @return - True if you are the owner, false otherwise
- * @throws OzoneException
- */
- public boolean checkVolumeAccess(String volume, OzoneAcl acl)
- throws OzoneException {
- lock.readLock().lock();
- try {
- byte[] volumeInfo =
- metadataDB.get(volume.getBytes(encoding));
- if (volumeInfo == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, null);
- }
-
- VolumeInfo info = VolumeInfo.parse(new String(volumeInfo, encoding));
- return info.getOwner().getName().equals(acl.getName());
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, null, ex);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * getVolumeInfo returns the Volume Info of a specific volume.
- *
- * @param args - Volume args
- * @return VolumeInfo
- * @throws OzoneException
- */
- public VolumeInfo getVolumeInfo(VolumeArgs args) throws OzoneException {
- lock.readLock().lock();
- try {
- byte[] volumeInfo =
- metadataDB.get(args.getVolumeName().getBytes(encoding));
- if (volumeInfo == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, args);
- }
-
- return VolumeInfo.parse(new String(volumeInfo, encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Returns all the volumes owned by a specific user.
- *
- * @param args - User Args
- * @return - ListVolumes
- * @throws OzoneException
- */
- public ListVolumes listVolumes(ListArgs args) throws OzoneException {
- lock.readLock().lock();
- try {
- if (args.isRootScan()) {
- return listAllVolumes(args);
- }
-
- UserArgs uArgs = (UserArgs) args.getArgs();
- byte[] volumeList = userDB.get(uArgs.getUserName().getBytes(encoding));
- if (volumeList == null) {
- throw ErrorTable.newError(ErrorTable.USER_NOT_FOUND, uArgs);
- }
-
- String prefix = args.getPrefix();
- int maxCount = args.getMaxKeys();
- String prevKey = args.getPrevKey();
- if (prevKey != null) {
- // Format is username/volumeName, in local mode we don't use the
- // user name since we have a userName DB.
- String[] volName = args.getPrevKey().split("/");
- if (volName.length < 2) {
- throw ErrorTable.newError(ErrorTable.USER_NOT_FOUND, uArgs);
- }
- prevKey = volName[1];
- }
- return getFilteredVolumes(volumeList, prefix, prevKey, maxCount);
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args.getArgs(), ex);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Returns a List of Volumes that meet the prefix, prevkey and maxCount
- * constraints.
- *
- * @param volumeList - Byte Array of Volume Info.
- * @param prefix - prefix string.
- * @param prevKey - PrevKey
- * @param maxCount - Maximum Count.
- * @return ListVolumes.
- * @throws IOException
- */
- private ListVolumes getFilteredVolumes(byte[] volumeList, String prefix,
- String prevKey, int maxCount) throws
- IOException {
- ListVolumes volumes = ListVolumes.parse(new String(volumeList,
- encoding));
- int currentCount = 0;
- ListIterator<VolumeInfo> iter = volumes.getVolumes().listIterator();
- ListVolumes filteredVolumes = new ListVolumes();
- while (currentCount < maxCount && iter.hasNext()) {
- VolumeInfo vInfo = iter.next();
- if (isMatchingPrefix(prefix, vInfo) && isAfterKey(prevKey, vInfo)) {
- filteredVolumes.addVolume(vInfo);
- currentCount++;
- }
- }
- return filteredVolumes;
- }
-
- /**
- * Returns all volumes in a cluster.
- *
- * @param args - ListArgs.
- * @return ListVolumes.
- * @throws OzoneException
- */
- public ListVolumes listAllVolumes(ListArgs args)
- throws OzoneException, IOException {
- String prefix = args.getPrefix();
- final String prevKey;
- int maxCount = args.getMaxKeys();
- String userName = null;
-
- if (args.getPrevKey() != null) {
- // Format is username/volumeName
- String[] volName = args.getPrevKey().split("/");
- if (volName.length < 2) {
- throw ErrorTable.newError(ErrorTable.USER_NOT_FOUND, args.getArgs());
- }
-
- byte[] userNameBytes = userDB.get(volName[0].getBytes(encoding));
- userName = new String(userNameBytes, encoding);
- prevKey = volName[1];
- } else {
- userName = new String(userDB.peekAround(0, null).getKey(), encoding);
- prevKey = null;
- }
-
- if (userName == null || userName.isEmpty()) {
- throw ErrorTable.newError(ErrorTable.USER_NOT_FOUND, args.getArgs());
- }
-
- ListVolumes returnSet = new ListVolumes();
- // we need to iterate through users until we get maxcount volumes
- // or no more volumes are left.
- userDB.iterate(null, (key, value) -> {
- int currentSize = returnSet.getVolumes().size();
- if (currentSize < maxCount) {
- String name = new String(key, encoding);
- byte[] volumeList = userDB.get(name.getBytes(encoding));
- if (volumeList == null) {
- throw new IOException(
- ErrorTable.newError(ErrorTable.USER_NOT_FOUND, args.getArgs()));
- }
- returnSet.getVolumes().addAll(
- getFilteredVolumes(volumeList, prefix, prevKey,
- maxCount - currentSize).getVolumes());
- return true;
- } else {
- return false;
- }
- });
-
- return returnSet;
- }
-
- /**
- * Checks if a name starts with a matching prefix.
- *
- * @param prefix - prefix string.
- * @param vInfo - volume info.
- * @return true or false.
- */
- private boolean isMatchingPrefix(String prefix, VolumeInfo vInfo) {
- if (prefix == null || prefix.isEmpty()) {
- return true;
- }
- return vInfo.getVolumeName().startsWith(prefix);
- }
-
- /**
- * Checks if the key is after the prevKey.
- *
- * @param prevKey - String prevKey.
- * @param vInfo - volume Info.
- * @return - true or false.
- */
- private boolean isAfterKey(String prevKey, VolumeInfo vInfo) {
- if (prevKey == null || prevKey.isEmpty()) {
- return true;
- }
- return prevKey.compareTo(vInfo.getVolumeName()) < 0;
- }
-
- /**
- * Deletes a volume if it exists and is empty.
- *
- * @param args - volume args
- * @throws OzoneException
- */
- public void deleteVolume(VolumeArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- byte[] volumeName =
- metadataDB.get(args.getVolumeName().getBytes(encoding));
- if (volumeName == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, args);
- }
-
- VolumeInfo vInfo = VolumeInfo.parse(new String(volumeName, encoding));
-
- // Only remove volumes if they are empty.
- if (vInfo.getBucketCount() > 0) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_EMPTY, args);
- }
-
- ListVolumes volumeList;
- String user = vInfo.getOwner().getName();
- byte[] userVolumes = userDB.get(user.getBytes(encoding));
- if (userVolumes == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, args);
- }
-
- volumeList = ListVolumes.parse(new String(userVolumes, encoding));
- volumeList.getVolumes().remove(vInfo);
-
- metadataDB.delete(args.getVolumeName().getBytes(encoding));
- userDB.put(user.getBytes(encoding),
- volumeList.toDBString().getBytes(encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Create a bucket if it does not exist.
- *
- * @param args - BucketArgs
- * @throws OzoneException
- */
- public void createBucket(BucketArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- // check if volume exists, buckets cannot be created without volumes
- byte[] volumeName = metadataDB.get(args.getVolumeName()
- .getBytes(encoding));
- if (volumeName == null) {
- throw ErrorTable.newError(ErrorTable.VOLUME_NOT_FOUND, args);
- }
-
- // A resource name is volume/bucket -- That is the key in metadata table
- byte[] bucketName = metadataDB.get(args.getResourceName()
- .getBytes(encoding));
- if (bucketName != null) {
- throw ErrorTable.newError(ErrorTable.BUCKET_ALREADY_EXISTS, args);
- }
-
- BucketInfo bucketInfo =
- new BucketInfo(args.getVolumeName(), args.getBucketName());
-
- if (args.getRemoveAcls() != null) {
- OzoneException ex = ErrorTable.newError(ErrorTable.MALFORMED_ACL, args);
- ex.setMessage("Remove ACLs specified in bucket create. Please remove "
- + "them and retry.");
- throw ex;
- }
-
- VolumeInfo volInfo = VolumeInfo.parse(new String(volumeName, encoding));
- volInfo.setBucketCount(volInfo.getBucketCount() + 1);
-
- bucketInfo.setAcls(args.getAddAcls());
- bucketInfo.setStorageType(args.getStorageType());
- bucketInfo.setVersioning(args.getVersioning());
- ListBuckets bucketList;
-
- // get bucket list from user/volume -> bucketList
- byte[] volumeBuckets = userDB.get(args.getParentName()
- .getBytes(encoding));
- if (volumeBuckets == null) {
- bucketList = new ListBuckets();
- } else {
- bucketList = ListBuckets.parse(new String(volumeBuckets, encoding));
- }
-
- bucketList.addBucket(bucketInfo);
- bucketList.sort();
-
- // Update Volume->bucketCount
- userDB.put(args.getVolumeName().getBytes(encoding),
- volInfo.toDBString().getBytes(encoding));
-
- // Now update the userDB with user/volume -> bucketList
- userDB.put(args.getParentName().getBytes(encoding),
- bucketList.toDBString().getBytes(encoding));
-
- // Update userDB with volume/bucket -> empty key list
- userDB.put(args.getResourceName().getBytes(encoding),
- new ListKeys().toDBString().getBytes(encoding));
-
- // and update the metadataDB with volume/bucket->BucketInfo
- metadataDB.put(args.getResourceName().getBytes(encoding),
- bucketInfo.toDBString().getBytes(encoding));
-
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Updates the Bucket properties like ACls and Storagetype.
- *
- * @param args - Bucket Args
- * @param property - Flag which tells us what property to upgrade
- * @throws OzoneException
- */
- public void setBucketProperty(BucketArgs args, BucketProperty property)
- throws OzoneException {
-
- lock.writeLock().lock();
- try {
- // volume/bucket-> bucketInfo
- byte[] bucketInfo = metadataDB.get(args.getResourceName().
- getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- BucketInfo info = BucketInfo.parse(new String(bucketInfo, encoding));
- byte[] volumeBuckets = userDB.get(args.getParentName()
- .getBytes(encoding));
- ListBuckets bucketList = ListBuckets.parse(new String(volumeBuckets,
- encoding));
- bucketList.getBuckets().remove(info);
-
- switch (property) {
- case ACLS:
- processRemoveAcls(args, info);
- processAddAcls(args, info);
- break;
- case STORAGETYPE:
- info.setStorageType(args.getStorageType());
- break;
- case VERSIONING:
- info.setVersioning(args.getVersioning());
- break;
- default:
- OzoneException ozEx =
- ErrorTable.newError(ErrorTable.BAD_PROPERTY, args);
- ozEx.setMessage("Bucket property is not recognized.");
- throw ozEx;
- }
-
- bucketList.addBucket(info);
- metadataDB.put(args.getResourceName().getBytes(encoding),
- info.toDBString().getBytes(encoding));
-
- userDB.put(args.getParentName().getBytes(encoding),
- bucketList.toDBString().getBytes(encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Process Remove Acls and remove them from the bucket.
- *
- * @param args - BucketArgs
- * @param info - BucketInfo
- */
- private void processRemoveAcls(BucketArgs args, BucketInfo info) {
- List<OzoneAcl> removeAcls = args.getRemoveAcls();
- if ((removeAcls == null) || (info.getAcls() == null)) {
- return;
- }
- for (OzoneAcl racl : args.getRemoveAcls()) {
- ListIterator<OzoneAcl> aclIter = info.getAcls().listIterator();
- while (aclIter.hasNext()) {
- if (racl.equals(aclIter.next())) {
- aclIter.remove();
- break;
- }
- }
- }
- }
-
- /**
- * Process Add Acls and Add them to the bucket.
- *
- * @param args - BucketArgs
- * @param info - BucketInfo
- */
- private void processAddAcls(BucketArgs args, BucketInfo info) {
- List<OzoneAcl> addAcls = args.getAddAcls();
- if ((addAcls == null)) {
- return;
- }
-
- if (info.getAcls() == null) {
- info.setAcls(addAcls);
- return;
- }
-
- for (OzoneAcl newacl : addAcls) {
- ListIterator<OzoneAcl> aclIter = info.getAcls().listIterator();
- while (aclIter.hasNext()) {
- if (newacl.equals(aclIter.next())) {
- continue;
- }
- }
- info.getAcls().add(newacl);
- }
- }
-
- /**
- * Deletes a given bucket.
- *
- * @param args - BucketArgs
- * @throws OzoneException
- */
- public void deleteBucket(BucketArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- byte[] bucketInfo = metadataDB.get(args.getResourceName()
- .getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- BucketInfo bInfo = BucketInfo.parse(new String(bucketInfo, encoding));
-
- // Only remove buckets if they are empty.
- if (bInfo.getKeyCount() > 0) {
- throw ErrorTable.newError(ErrorTable.BUCKET_NOT_EMPTY, args);
- }
-
- byte[] bucketBytes = userDB.get(args.getParentName().getBytes(encoding));
- if (bucketBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- ListBuckets bucketList =
- ListBuckets.parse(new String(bucketBytes, encoding));
- bucketList.getBuckets().remove(bInfo);
-
- metadataDB.delete(args.getResourceName().getBytes(encoding));
- userDB.put(args.getParentName().getBytes(encoding),
- bucketList.toDBString().getBytes(encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Returns the Bucket info for a given bucket.
- *
- * @param args - Bucket Args
- * @return BucketInfo - Bucket Information
- * @throws OzoneException
- */
- public BucketInfo getBucketInfo(BucketArgs args) throws OzoneException {
- lock.readLock().lock();
- try {
- byte[] bucketBytes = metadataDB.get(args.getResourceName()
- .getBytes(encoding));
- if (bucketBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- return BucketInfo.parse(new String(bucketBytes, encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, ex);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Returns a list of buckets for a given volume.
- *
- * @param args - volume args
- * @return List of buckets
- * @throws OzoneException
- */
- public ListBuckets listBuckets(ListArgs args) throws OzoneException {
- lock.readLock().lock();
- try {
- Preconditions.checkState(args.getArgs() instanceof VolumeArgs);
- VolumeArgs vArgs = (VolumeArgs) args.getArgs();
- String userVolKey = vArgs.getUserName() + "/" + vArgs.getVolumeName();
-
- // TODO : Query using Prefix and PrevKey
- byte[] bucketBytes = userDB.get(userVolKey.getBytes(encoding));
- if (bucketBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_VOLUME_NAME,
- args.getArgs());
- }
- return ListBuckets.parse(new String(bucketBytes, encoding));
- } catch (IOException ex) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args.getArgs(), ex);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Creates a key and returns a stream to which this key can be written to.
- *
- * @param args KeyArgs
- * @return - A stream into which key can be written to.
- * @throws OzoneException
- */
- public OutputStream createKey(KeyArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- String fileNameHash = DigestUtils.sha256Hex(args.getResourceName());
-
- // Please don't try trillion objects unless the physical file system
- // is capable of doing that in a single directory.
-
- String fullPath = storageRoot + OBJECT_DIR + fileNameHash;
- File f = new File(fullPath);
-
- // In real ozone it would not be this way, a file will be overwritten
- // only if the upload is successful.
- if (f.exists()) {
- LOG.debug("we are overwriting a file. This is by design.");
- if (!f.delete()) {
- LOG.error("Unable to delete the file: {}", fullPath);
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args);
- }
- }
-
- // f.createNewFile();
- FileOutputStream fsStream = new FileOutputStream(f);
- inProgressObjects.put(fsStream, fullPath);
-
- return fsStream;
- } catch (IOException e) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, e);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * commit keys moves an In progress object into the metadata store so that key
- * is visible in the metadata operations from that point onwards.
- *
- * @param args Object args
- * @throws OzoneException
- */
- public void commitKey(KeyArgs args, OutputStream stream)
- throws OzoneException {
- SimpleDateFormat format =
- new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss ZZZ", Locale.US);
- lock.writeLock().lock();
-
- try {
- byte[] bucketInfo = metadataDB.get(args.getParentName()
- .getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_RESOURCE_NAME, args);
- }
- BucketInfo bInfo = BucketInfo.parse(new String(bucketInfo, encoding));
- bInfo.setKeyCount(bInfo.getKeyCount() + 1);
-
- String fileNameHash = inProgressObjects.get(stream);
- inProgressObjects.remove(stream);
- if (fileNameHash == null) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args);
- }
-
- ListKeys keyList;
- byte[] bucketListBytes = userDB.get(args.getParentName()
- .getBytes(encoding));
- keyList = ListKeys.parse(new String(bucketListBytes, encoding));
- KeyInfo keyInfo;
-
- byte[] objectBytes = metadataDB.get(args.getResourceName()
- .getBytes(encoding));
-
- if (objectBytes != null) {
- // we are overwriting an existing object.
- // TODO : Emit info for Accounting
- keyInfo = KeyInfo.parse(new String(objectBytes, encoding));
- keyList.getKeyList().remove(keyInfo);
- } else {
- keyInfo = new KeyInfo();
- }
-
- keyInfo.setCreatedOn(format.format(new Date(System.currentTimeMillis())));
-
- // TODO : support version, we need to check if versioning
- // is switched on the bucket and make appropriate calls.
- keyInfo.setVersion(0);
-
- keyInfo.setDataFileName(fileNameHash);
- keyInfo.setKeyName(args.getKeyName());
- keyInfo.setMd5hash(args.getHash());
- keyInfo.setSize(args.getSize());
-
- keyList.getKeyList().add(keyInfo);
-
- // if the key exists, we overwrite happily :). since the
- // earlier call - createObject - has overwritten the data.
-
- metadataDB.put(args.getResourceName().getBytes(encoding),
- keyInfo.toDBString().getBytes(encoding));
-
- metadataDB.put(args.getParentName().getBytes(encoding),
- bInfo.toDBString().getBytes(encoding));
-
- userDB.put(args.getParentName().getBytes(encoding),
- keyList.toDBString().getBytes(encoding));
-
- } catch (IOException e) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, e);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * deletes an key from a given bucket.
- *
- * @param args - ObjectArgs
- * @throws OzoneException
- */
- public void deleteKey(KeyArgs args) throws OzoneException {
- lock.writeLock().lock();
- try {
- byte[] bucketInfo = metadataDB.get(args.getParentName()
- .getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
- BucketInfo bInfo = BucketInfo.parse(new String(bucketInfo, encoding));
- bInfo.setKeyCount(bInfo.getKeyCount() - 1);
-
-
- byte[] bucketListBytes = userDB.get(args.getParentName()
- .getBytes(encoding));
- if (bucketListBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
- ListKeys keyList = ListKeys.parse(new String(bucketListBytes, encoding));
-
-
- byte[] objectBytes = metadataDB.get(args.getResourceName()
- .getBytes(encoding));
- if (objectBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_KEY, args);
- }
-
- KeyInfo oInfo = KeyInfo.parse(new String(objectBytes, encoding));
- keyList.getKeyList().remove(oInfo);
-
- String fileNameHash = DigestUtils.sha256Hex(args.getResourceName());
-
- String fullPath = storageRoot + OBJECT_DIR + fileNameHash;
- File f = new File(fullPath);
-
- if (f.exists()) {
- if (!f.delete()) {
- throw ErrorTable.newError(ErrorTable.KEY_OPERATION_CONFLICT, args);
- }
- } else {
- throw ErrorTable.newError(ErrorTable.INVALID_KEY, args);
- }
-
-
- metadataDB.delete(args.getResourceName().getBytes(encoding));
- metadataDB.put(args.getParentName().getBytes(encoding),
- bInfo.toDBString().getBytes(encoding));
- userDB.put(args.getParentName().getBytes(encoding),
- keyList.toDBString().getBytes(encoding));
- } catch (IOException e) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, e);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- /**
- * Returns a Stream for the file.
- *
- * @param args - Object args
- * @return Stream
- * @throws IOException
- * @throws OzoneException
- */
- public LengthInputStream newKeyReader(KeyArgs args)
- throws IOException, OzoneException {
- lock.readLock().lock();
- try {
- String fileNameHash = DigestUtils.sha256Hex(args.getResourceName());
- String fullPath = storageRoot + OBJECT_DIR + fileNameHash;
- File f = new File(fullPath);
- if (!f.exists()) {
- throw ErrorTable.newError(ErrorTable.INVALID_RESOURCE_NAME, args);
- }
- long size = f.length();
-
- FileInputStream fileStream = new FileInputStream(f);
- return new LengthInputStream(fileStream, size);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Returns keys in a bucket.
- *
- * @param args
- * @return List of keys.
- * @throws IOException
- * @throws OzoneException
- */
- public ListKeys listKeys(ListArgs args) throws IOException, OzoneException {
- lock.readLock().lock();
- // TODO : Support Prefix and PrevKey lookup.
- try {
- Preconditions.checkState(args.getArgs() instanceof BucketArgs);
- BucketArgs bArgs = (BucketArgs) args.getArgs();
- byte[] bucketInfo = metadataDB.get(bArgs.getResourceName()
- .getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_RESOURCE_NAME, bArgs);
- }
-
- byte[] bucketListBytes = userDB.get(bArgs.getResourceName()
- .getBytes(encoding));
- if (bucketListBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_RESOURCE_NAME, bArgs);
- }
- return ListKeys.parse(new String(bucketListBytes, encoding));
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * Get the Key information for a given key.
- *
- * @param args - Key Args
- * @return KeyInfo - Key Information
- * @throws OzoneException
- */
- public KeyInfo getKeyInfo(KeyArgs args) throws IOException, OzoneException {
- lock.readLock().lock();
- try {
- byte[] bucketInfo = metadataDB
- .get(args.getParentName().getBytes(encoding));
- if (bucketInfo == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- byte[] bucketListBytes = userDB
- .get(args.getParentName().getBytes(encoding));
- if (bucketListBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_BUCKET_NAME, args);
- }
-
- byte[] objectBytes = metadataDB
- .get(args.getResourceName().getBytes(encoding));
- if (objectBytes == null) {
- throw ErrorTable.newError(ErrorTable.INVALID_KEY, args);
- }
-
- return KeyInfo.parse(new String(objectBytes, encoding));
- } catch (IOException e) {
- throw ErrorTable.newError(ErrorTable.SERVER_ERROR, args, e);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
- * This is used in updates to volume metadata.
- */
- public enum VolumeProperty {
- OWNER, QUOTA
- }
-
- /**
- * Bucket Properties.
- */
- public enum BucketProperty {
- ACLS, STORAGETYPE, VERSIONING
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/package-info.java
deleted file mode 100644
index 6bf6643..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/localstorage/package-info.java
+++ /dev/null
@@ -1,18 +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.web.localstorage;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/LengthInputStreamMessageBodyWriter.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/LengthInputStreamMessageBodyWriter.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/LengthInputStreamMessageBodyWriter.java
deleted file mode 100644
index 6db49d5..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/LengthInputStreamMessageBodyWriter.java
+++ /dev/null
@@ -1,59 +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.web.messages;
-
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ozone.client.io.LengthInputStream;
-
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.MultivaluedMap;
-import javax.ws.rs.ext.MessageBodyWriter;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Type;
-
-/**
- * Writes outbound HTTP response object bytes. The content length is determined
- * from the {@link LengthInputStream}.
- */
-public final class LengthInputStreamMessageBodyWriter
- implements MessageBodyWriter<LengthInputStream> {
- private static final int CHUNK_SIZE = 8192;
-
- @Override
- public long getSize(LengthInputStream lis, Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType) {
- return lis.getLength();
- }
-
- @Override
- public boolean isWriteable(Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType) {
- return LengthInputStream.class.isAssignableFrom(type);
- }
-
- @Override
- public void writeTo(LengthInputStream lis, Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType,
- MultivaluedMap<String, Object> httpHeaders,
- OutputStream out) throws IOException {
- IOUtils.copyBytes(lis, out, CHUNK_SIZE);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/StringMessageBodyWriter.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/StringMessageBodyWriter.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/StringMessageBodyWriter.java
deleted file mode 100644
index ad637af..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/StringMessageBodyWriter.java
+++ /dev/null
@@ -1,62 +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.web.messages;
-
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ozone.web.utils.OzoneUtils;
-
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.MultivaluedMap;
-import javax.ws.rs.ext.MessageBodyWriter;
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Type;
-
-/**
- * Writes outbound HTTP response strings. We use this rather than the built-in
- * writer so that we can determine content length from the string length instead
- * of possibly falling back to a chunked response.
- */
-public final class StringMessageBodyWriter implements
- MessageBodyWriter<String> {
- private static final int CHUNK_SIZE = 8192;
-
- @Override
- public long getSize(String str, Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType) {
- return str.length();
- }
-
- @Override
- public boolean isWriteable(Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType) {
- return String.class.isAssignableFrom(type);
- }
-
- @Override
- public void writeTo(String str, Class<?> type, Type genericType,
- Annotation[] annotations, MediaType mediaType,
- MultivaluedMap<String, Object> httpHeaders,
- OutputStream out) throws IOException {
- IOUtils.copyBytes(new ByteArrayInputStream(
- str.getBytes(OzoneUtils.ENCODING)), out, CHUNK_SIZE);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/package-info.java
deleted file mode 100644
index 273b3f5..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/messages/package-info.java
+++ /dev/null
@@ -1,18 +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.web.messages;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/CloseableCleanupListener.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/CloseableCleanupListener.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/CloseableCleanupListener.java
deleted file mode 100644
index 3d9db20..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/CloseableCleanupListener.java
+++ /dev/null
@@ -1,46 +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.web.netty;
-
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import org.apache.hadoop.io.IOUtils;
-
-import java.io.Closeable;
-
-/**
- * A {@link ChannelFutureListener} that closes {@link Closeable} resources.
- */
-final class CloseableCleanupListener implements ChannelFutureListener {
-
- private final Closeable[] closeables;
-
- /**
- * Creates a new CloseableCleanupListener.
- *
- * @param closeables any number of closeable resources
- */
- CloseableCleanupListener(Closeable... closeables) {
- this.closeables = closeables;
- }
-
- @Override
- public void operationComplete(ChannelFuture future) {
- IOUtils.cleanupWithLogger(null, closeables);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreChannelHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreChannelHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreChannelHandler.java
deleted file mode 100644
index 89c196c..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreChannelHandler.java
+++ /dev/null
@@ -1,78 +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.web.netty;
-
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
-import static io.netty.handler.codec.http.HttpHeaders.Values.CLOSE;
-import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
-import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-
-/**
- * Abstract base class for the multiple Netty channel handlers used in the
- * Object Store Netty channel pipeline.
- */
-abstract class ObjectStoreChannelHandler<T>
- extends SimpleChannelInboundHandler<T> {
-
- /** Log usable in all subclasses. */
- protected static final Logger LOG =
- LoggerFactory.getLogger(ObjectStoreChannelHandler.class);
-
- /**
- * Handles uncaught exceptions in the channel pipeline by sending an internal
- * server error response if the channel is still active.
- *
- * @param ctx ChannelHandlerContext to receive response
- * @param cause Throwable that was unhandled in the channel pipeline
- */
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- LOG.error("Unexpected exception in Netty pipeline.", cause);
- if (ctx.channel().isActive()) {
- sendErrorResponse(ctx, INTERNAL_SERVER_ERROR);
- }
- }
-
- /**
- * Sends an error response. This method is used when an unexpected error is
- * encountered within the channel pipeline, outside of the actual Object Store
- * application. It always closes the connection, because we can't in general
- * know the state of the connection when these errors occur, so attempting to
- * keep the connection alive could be unpredictable.
- *
- * @param ctx ChannelHandlerContext to receive response
- * @param status HTTP response status
- */
- protected static void sendErrorResponse(ChannelHandlerContext ctx,
- HttpResponseStatus status) {
- HttpResponse nettyResp = new DefaultFullHttpResponse(HTTP_1_1, status);
- nettyResp.headers().set(CONTENT_LENGTH, 0);
- nettyResp.headers().set(CONNECTION, CLOSE);
- ctx.writeAndFlush(nettyResp).addListener(ChannelFutureListener.CLOSE);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainer.java
deleted file mode 100644
index c7b516f..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainer.java
+++ /dev/null
@@ -1,348 +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.web.netty;
-
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
-import static io.netty.handler.codec.http.HttpHeaders.Names.TRANSFER_ENCODING;
-import static io.netty.handler.codec.http.HttpHeaders.Names.HOST;
-import static io.netty.handler.codec.http.HttpHeaders.Values.KEEP_ALIVE;
-import static io.netty.handler.codec.http.HttpHeaders.Values.CLOSE;
-import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.TimeUnit;
-
-import com.sun.jersey.core.header.InBoundHeaders;
-import com.sun.jersey.spi.container.ContainerRequest;
-import com.sun.jersey.spi.container.ContainerResponse;
-import com.sun.jersey.spi.container.ContainerResponseWriter;
-import com.sun.jersey.spi.container.WebApplication;
-
-import io.netty.handler.codec.http.DefaultHttpResponse;
-//import io.netty.handler.codec.http.HttpUtil;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ozone.web.interfaces.StorageHandler;
-import org.apache.hadoop.ozone.web.handlers.StorageHandlerBuilder;
-
-/**
- * This is a custom Jersey container that hosts the Object Store web
- * application. It supports dispatching an inbound Netty {@link HttpRequest}
- * to the Object Store Jersey application. Request dispatching must run
- * asynchronously, because the Jersey application must consume the inbound
- * HTTP request from a piped stream and produce the outbound HTTP response
- * for another piped stream.The Netty channel handlers consume the connected
- * ends of these piped streams. Request dispatching cannot run directly on
- * the Netty threads, or there would be a risk of deadlock (one thread
- * producing/consuming its end of the pipe while no other thread is
- * producing/consuming the opposite end).
- */
-public final class ObjectStoreJerseyContainer {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(ObjectStoreJerseyContainer.class);
-
- private final WebApplication webapp;
-
- private StorageHandler storageHandler;
-
- /**
- * Creates a new ObjectStoreJerseyContainer.
- *
- * @param webapp web application
- */
- public ObjectStoreJerseyContainer(WebApplication webapp) {
- this.webapp = webapp;
- }
-
- /**
- * Sets the {@link StorageHandler}. This must be called before dispatching any
- * requests.
- *
- * @param newStorageHandler {@link StorageHandler} implementation
- */
- public void setStorageHandler(StorageHandler newStorageHandler) {
- this.storageHandler = newStorageHandler;
- }
-
- /**
- * Asynchronously executes an HTTP request.
- *
- * @param nettyReq HTTP request
- * @param reqIn input stream for reading request body
- * @param respOut output stream for writing response body
- */
- public Future<HttpResponse> dispatch(HttpRequest nettyReq, InputStream reqIn,
- OutputStream respOut) {
- // The request executes on a separate background thread. As soon as enough
- // processing has completed to bootstrap the outbound response, the thread
- // counts down on a latch. This latch also unblocks callers trying to get
- // the asynchronous response out of the returned future.
- final CountDownLatch latch = new CountDownLatch(1);
- final RequestRunner runner = new RequestRunner(nettyReq, reqIn, respOut,
- latch);
- final Thread thread = new Thread(runner);
- thread.setDaemon(true);
- thread.start();
- return new Future<HttpResponse>() {
-
- private volatile boolean isCancelled = false;
-
- @Override
- public boolean cancel(boolean mayInterruptIfRunning) {
- if (latch.getCount() == 0) {
- return false;
- }
- if (!mayInterruptIfRunning) {
- return false;
- }
- if (!thread.isAlive()) {
- return false;
- }
- thread.interrupt();
- try {
- thread.join();
- } catch (InterruptedException e) {
- LOG.info("Interrupted while attempting to cancel dispatch thread.");
- Thread.currentThread().interrupt();
- return false;
- }
- isCancelled = true;
- return true;
- }
-
- @Override
- public HttpResponse get()
- throws InterruptedException, ExecutionException {
- checkCancelled();
- latch.await();
- return this.getOrThrow();
- }
-
- @Override
- public HttpResponse get(long timeout, TimeUnit unit)
- throws InterruptedException, ExecutionException, TimeoutException {
- checkCancelled();
- if (!latch.await(timeout, unit)) {
- throw new TimeoutException(String.format(
- "Timed out waiting for HttpResponse after %d %s.",
- timeout, unit.toString().toLowerCase()));
- }
- return this.getOrThrow();
- }
-
- @Override
- public boolean isCancelled() {
- return isCancelled;
- }
-
- @Override
- public boolean isDone() {
- return !isCancelled && latch.getCount() == 0;
- }
-
- private void checkCancelled() {
- if (isCancelled()) {
- throw new CancellationException();
- }
- }
-
- private HttpResponse getOrThrow() throws ExecutionException {
- try {
- return runner.getResponse();
- } catch (Exception e) {
- throw new ExecutionException(e);
- }
- }
- };
- }
-
- /**
- * Runs the actual handling of the HTTP request.
- */
- private final class RequestRunner implements Runnable,
- ContainerResponseWriter {
-
- private final CountDownLatch latch;
- private final HttpRequest nettyReq;
- private final InputStream reqIn;
- private final OutputStream respOut;
-
- private Exception exception;
- private HttpResponse nettyResp;
-
- /**
- * Creates a new RequestRunner.
- *
- * @param nettyReq HTTP request
- * @param reqIn input stream for reading request body
- * @param respOut output stream for writing response body
- * @param latch for coordinating asynchronous return of HTTP response
- */
- RequestRunner(HttpRequest nettyReq, InputStream reqIn,
- OutputStream respOut, CountDownLatch latch) {
- this.latch = latch;
- this.nettyReq = nettyReq;
- this.reqIn = reqIn;
- this.respOut = respOut;
- }
-
- @Override
- public void run() {
- LOG.trace("begin RequestRunner, nettyReq = {}", this.nettyReq);
- StorageHandlerBuilder.setStorageHandler(
- ObjectStoreJerseyContainer.this.storageHandler);
- try {
- ContainerRequest jerseyReq = nettyRequestToJerseyRequest(
- ObjectStoreJerseyContainer.this.webapp, this.nettyReq, this.reqIn);
- ObjectStoreJerseyContainer.this.webapp.handleRequest(jerseyReq, this);
- } catch (Exception e) {
- LOG.error("Error running Jersey Request Runner", e);
- this.exception = e;
- this.latch.countDown();
- } finally {
- IOUtils.cleanupWithLogger(null, this.reqIn, this.respOut);
- StorageHandlerBuilder.removeStorageHandler();
- }
- LOG.trace("end RequestRunner, nettyReq = {}", this.nettyReq);
- }
-
- /**
- * This is a callback triggered by Jersey as soon as dispatch has completed
- * to the point of knowing what kind of response to return. We save the
- * response and trigger the latch to unblock callers waiting on the
- * asynchronous return of the response. Our response always sets a
- * Content-Length header. (We do not support Transfer-Encoding: chunked.)
- * We also return the output stream for Jersey to use for writing the
- * response body.
- *
- * @param contentLength length of response
- * @param jerseyResp HTTP response returned by Jersey
- * @return OutputStream for Jersey to use for writing the response body
- */
- @Override
- public OutputStream writeStatusAndHeaders(long contentLength,
- ContainerResponse jerseyResp) {
- LOG.trace(
- "begin writeStatusAndHeaders, contentLength = {}, jerseyResp = {}.",
- contentLength, jerseyResp);
- this.nettyResp = jerseyResponseToNettyResponse(jerseyResp);
- this.nettyResp.headers().set(CONTENT_LENGTH, Math.max(0, contentLength));
- this.nettyResp.headers().set(CONNECTION,
- HttpHeaders.isKeepAlive(this.nettyReq) ? KEEP_ALIVE : CLOSE);
- this.latch.countDown();
- LOG.trace(
- "end writeStatusAndHeaders, contentLength = {}, jerseyResp = {}.",
- contentLength, jerseyResp);
- return this.respOut;
- }
-
- /**
- * This is a callback triggered by Jersey after it has completed writing the
- * response body to the stream. We must close the stream here to unblock
- * the Netty thread consuming the last chunk of the response from the input
- * end of the piped stream.
- *
- * @throws IOException if there is an I/O error
- */
- @Override
- public void finish() throws IOException {
- IOUtils.cleanupWithLogger(null, this.respOut);
- }
-
- /**
- * Gets the HTTP response calculated by the Jersey application, or throws an
- * exception if an error occurred during processing. It only makes sense to
- * call this method after waiting on the latch to trigger.
- *
- * @return HTTP response
- * @throws Exception if there was an error executing the request
- */
- public HttpResponse getResponse() throws Exception {
- if (this.exception != null) {
- throw this.exception;
- }
- return this.nettyResp;
- }
- }
-
- /**
- * Converts a Jersey HTTP response object to a Netty HTTP response object.
- *
- * @param jerseyResp Jersey HTTP response
- * @return Netty HTTP response
- */
- private static HttpResponse jerseyResponseToNettyResponse(
- ContainerResponse jerseyResp) {
- HttpResponse nettyResp = new DefaultHttpResponse(HTTP_1_1,
- HttpResponseStatus.valueOf(jerseyResp.getStatus()));
- for (Map.Entry<String, List<Object>> header :
- jerseyResp.getHttpHeaders().entrySet()) {
- if (!header.getKey().equalsIgnoreCase(CONTENT_LENGTH.toString()) &&
- !header.getKey().equalsIgnoreCase(TRANSFER_ENCODING.toString())) {
- nettyResp.headers().set(header.getKey(), header.getValue());
- }
- }
- return nettyResp;
- }
-
- /**
- * Converts a Netty HTTP request object to a Jersey HTTP request object.
- *
- * @param webapp web application
- * @param nettyReq Netty HTTP request
- * @param reqIn input stream for reading request body
- * @return Jersey HTTP request
- * @throws URISyntaxException if there is an error handling the request URI
- */
- private static ContainerRequest nettyRequestToJerseyRequest(
- WebApplication webapp, HttpRequest nettyReq, InputStream reqIn)
- throws URISyntaxException {
- HttpHeaders nettyHeaders = nettyReq.headers();
- InBoundHeaders jerseyHeaders = new InBoundHeaders();
- for (String name : nettyHeaders.names()) {
- jerseyHeaders.put(name, nettyHeaders.getAll(name));
- }
- String host = nettyHeaders.get(HOST);
- String scheme = host.startsWith("https") ? "https://" : "http://";
- String baseUri = scheme + host + "/";
- String reqUri = scheme + host + nettyReq.getUri();
- LOG.trace("baseUri = {}, reqUri = {}", baseUri, reqUri);
- return new ContainerRequest(webapp, nettyReq.getMethod().name(),
- new URI(baseUri), new URI(reqUri), jerseyHeaders, reqIn);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainerProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainerProvider.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainerProvider.java
deleted file mode 100644
index e943969..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/ObjectStoreJerseyContainerProvider.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.web.netty;
-
-import com.sun.jersey.api.container.ContainerException;
-import com.sun.jersey.api.core.ResourceConfig;
-import com.sun.jersey.spi.container.ContainerProvider;
-import com.sun.jersey.spi.container.WebApplication;
-
-/**
- * This is a Jersey {@link ContainerProvider} capable of boostrapping the
- * Object Store web application into a custom container. It must be registered
- * using the Java service loader mechanism by listing it in
- * META-INF/services/com.sun.jersey.spi.container.ContainerProvider .
- */
-public final class ObjectStoreJerseyContainerProvider
- implements ContainerProvider<ObjectStoreJerseyContainer> {
-
- @Override
- public ObjectStoreJerseyContainer createContainer(
- Class<ObjectStoreJerseyContainer> type, ResourceConfig conf,
- WebApplication webapp) throws ContainerException {
- return new ObjectStoreJerseyContainer(webapp);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestContentObjectStoreChannelHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestContentObjectStoreChannelHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestContentObjectStoreChannelHandler.java
deleted file mode 100644
index 0a2f22d..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestContentObjectStoreChannelHandler.java
+++ /dev/null
@@ -1,117 +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.web.netty;
-
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.HttpContent;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.LastHttpContent;
-import io.netty.handler.stream.ChunkedStream;
-import org.apache.hadoop.io.IOUtils;
-
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.concurrent.Future;
-
-/**
- * Object Store Netty channel pipeline handler that handles inbound
- * {@link HttpContent} fragments for the request body by sending the bytes into
- * the pipe so that the application dispatch thread can read it.
- * After receiving the {@link LastHttpContent}, this handler also flushes the
- * response.
- */
-public final class RequestContentObjectStoreChannelHandler
- extends ObjectStoreChannelHandler<HttpContent> {
-
- private final HttpRequest nettyReq;
- private final Future<HttpResponse> nettyResp;
- private final OutputStream reqOut;
- private final InputStream respIn;
- private ObjectStoreJerseyContainer jerseyContainer;
-
- /**
- * Creates a new RequestContentObjectStoreChannelHandler.
- *
- * @param nettyReq HTTP request
- * @param nettyResp asynchronous HTTP response
- * @param reqOut output stream for writing request body
- * @param respIn input stream for reading response body
- * @param jerseyContainer jerseyContainer to handle the request
- */
- public RequestContentObjectStoreChannelHandler(HttpRequest nettyReq,
- Future<HttpResponse> nettyResp, OutputStream reqOut, InputStream respIn,
- ObjectStoreJerseyContainer jerseyContainer) {
- this.nettyReq = nettyReq;
- this.nettyResp = nettyResp;
- this.reqOut = reqOut;
- this.respIn = respIn;
- this.jerseyContainer = jerseyContainer;
- }
-
- @Override
- public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
- ctx.flush();
- }
-
- @Override
- public void channelRead0(ChannelHandlerContext ctx, HttpContent content)
- throws Exception {
- LOG.trace(
- "begin RequestContentObjectStoreChannelHandler channelRead0, " +
- "ctx = {}, content = {}", ctx, content);
- content.content().readBytes(this.reqOut, content.content().readableBytes());
- if (content instanceof LastHttpContent) {
- IOUtils.cleanupWithLogger(null, this.reqOut);
- ctx.write(this.nettyResp.get());
- ChannelFuture respFuture = ctx.writeAndFlush(new ChunkedStream(
- this.respIn));
- respFuture.addListener(new CloseableCleanupListener(this.respIn));
- if (!HttpHeaders.isKeepAlive(this.nettyReq)) {
- respFuture.addListener(ChannelFutureListener.CLOSE);
- } else {
- respFuture.addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- // Notify client this is the last content for current request.
- ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
- // Reset the pipeline handler for next request to reuses the
- // same connection.
- RequestDispatchObjectStoreChannelHandler h =
- new RequestDispatchObjectStoreChannelHandler(jerseyContainer);
- ctx.pipeline().replace(ctx.pipeline().last(),
- RequestDispatchObjectStoreChannelHandler.class.getSimpleName(),
- h);
- }
- });
- }
- }
- LOG.trace(
- "end RequestContentObjectStoreChannelHandler channelRead0, " +
- "ctx = {}, content = {}", ctx, content);
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- super.exceptionCaught(ctx, cause);
- IOUtils.cleanupWithLogger(null, this.reqOut, this.respIn);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestDispatchObjectStoreChannelHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestDispatchObjectStoreChannelHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestDispatchObjectStoreChannelHandler.java
deleted file mode 100644
index add827a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/RequestDispatchObjectStoreChannelHandler.java
+++ /dev/null
@@ -1,103 +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.web.netty;
-
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-import org.apache.hadoop.io.IOUtils;
-
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
-import java.util.concurrent.Future;
-
-import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
-import static io.netty.handler.codec.http.HttpResponseStatus.CONTINUE;
-import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-
-/**
- * Object Store Netty channel pipeline handler that handles an inbound
- * {@link HttpRequest} by dispatching it to the Object Store Jersey container.
- * The handler establishes 2 sets of connected piped streams: one for inbound
- * request handling and another for outbound response handling. The relevant
- * ends of these pipes are handed off to the Jersey application dispatch and the
- * next channel handler, which is responsible for streaming in the inbound
- * request body and flushing out the response body.
- */
-public final class RequestDispatchObjectStoreChannelHandler
- extends ObjectStoreChannelHandler<HttpRequest> {
-
- private final ObjectStoreJerseyContainer jerseyContainer;
-
- private PipedInputStream reqIn;
- private PipedOutputStream reqOut;
- private PipedInputStream respIn;
- private PipedOutputStream respOut;
-
- /**
- * Creates a new RequestDispatchObjectStoreChannelHandler.
- *
- * @param jerseyContainer Object Store application Jersey container for
- * request dispatch
- */
- public RequestDispatchObjectStoreChannelHandler(
- ObjectStoreJerseyContainer jerseyContainer) {
- this.jerseyContainer = jerseyContainer;
- }
-
- @Override
- public void channelRead0(ChannelHandlerContext ctx, HttpRequest nettyReq)
- throws Exception {
- LOG.trace("begin RequestDispatchObjectStoreChannelHandler channelRead0, " +
- "ctx = {}, nettyReq = {}", ctx, nettyReq);
- if (!nettyReq.getDecoderResult().isSuccess()) {
- sendErrorResponse(ctx, BAD_REQUEST);
- return;
- }
-
- this.reqIn = new PipedInputStream();
- this.reqOut = new PipedOutputStream(reqIn);
- this.respIn = new PipedInputStream();
- this.respOut = new PipedOutputStream(respIn);
-
- if (HttpHeaders.is100ContinueExpected(nettyReq)) {
- LOG.trace("Sending continue response.");
- ctx.writeAndFlush(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
- }
-
- Future<HttpResponse> nettyResp = this.jerseyContainer.dispatch(nettyReq,
- reqIn, respOut);
-
- ctx.pipeline().replace(this,
- RequestContentObjectStoreChannelHandler.class.getSimpleName(),
- new RequestContentObjectStoreChannelHandler(nettyReq, nettyResp,
- reqOut, respIn, jerseyContainer));
-
- LOG.trace("end RequestDispatchObjectStoreChannelHandler channelRead0, " +
- "ctx = {}, nettyReq = {}", ctx, nettyReq);
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- super.exceptionCaught(ctx, cause);
- IOUtils.cleanupWithLogger(null, this.reqIn, this.reqOut, this.respIn,
- this.respOut);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/package-info.java
deleted file mode 100644
index f4aa675..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/netty/package-info.java
+++ /dev/null
@@ -1,26 +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.
- */
-
-/**
- * Netty-based HTTP server implementation for Ozone.
- */
-@InterfaceAudience.Private
-package org.apache.hadoop.ozone.web.netty;
-
-import org.apache.hadoop.classification.InterfaceAudience;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/ozShell/Handler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/ozShell/Handler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/ozShell/Handler.java
deleted file mode 100644
index 015d46e..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/ozShell/Handler.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
- *
- * 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.web.ozShell;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.hadoop.ozone.web.client.OzoneRestClient;
-import org.apache.hadoop.ozone.web.client.OzoneRestClientException;
-import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.http.client.utils.URIBuilder;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-
-/**
- * Common interface for command handling.
- */
-public abstract class Handler {
- protected OzoneRestClient client;
-
- /**
- * Constructs a client object.
- */
- public Handler() {
- client = new OzoneRestClient();
- }
-
- /**
- * Executes the Client command.
- *
- * @param cmd - CommandLine
- * @throws IOException
- * @throws OzoneException
- * @throws URISyntaxException
- */
- protected abstract void execute(CommandLine cmd)
- throws IOException, OzoneException, URISyntaxException;
-
- /**
- * verifies user provided URI.
- *
- * @param uri - UriString
- * @return URI
- * @throws URISyntaxException
- * @throws OzoneException
- */
- protected URI verifyURI(String uri) throws URISyntaxException,
- OzoneException {
- if ((uri == null) || uri.isEmpty()) {
- throw new OzoneRestClientException(
- "Ozone URI is needed to execute this command.");
- }
- URIBuilder ozoneURI = new URIBuilder(uri);
-
- if (ozoneURI.getPort() == 0) {
- ozoneURI.setPort(Shell.DEFAULT_OZONE_PORT);
- }
- return ozoneURI.build();
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org