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 om...@apache.org on 2018/04/26 21:01:15 UTC
[46/68] [abbrv] [partial] hadoop git commit: HDFS-13405. Ozone:
Rename HDSL to HDDS. Contributed by Ajay Kumar, Elek Marton, Mukul Kumar Singh,
Shashikant Banerjee and Anu Engineer.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/package-info.java
new file mode 100644
index 0000000..fa5df11
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/package-info.java
@@ -0,0 +1,23 @@
+/**
+ * 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.common.helpers;
+
+/**
+ * Helper classes for the container protocol communication.
+ */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/Lease.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/Lease.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/Lease.java
new file mode 100644
index 0000000..dfa9315
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/Lease.java
@@ -0,0 +1,189 @@
+/**
+ * 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.lease;
+
+import org.apache.hadoop.util.Time;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+/**
+ * This class represents the lease created on a resource. Callback can be
+ * registered on the lease which will be executed in case of timeout.
+ *
+ * @param <T> Resource type for which the lease can be associated
+ */
+public class Lease<T> {
+
+ /**
+ * The resource for which this lease is created.
+ */
+ private final T resource;
+
+ private final long creationTime;
+
+ /**
+ * Lease lifetime in milliseconds.
+ */
+ private volatile long leaseTimeout;
+
+ private boolean expired;
+
+ /**
+ * Functions to be called in case of timeout.
+ */
+ private List<Callable<Void>> callbacks;
+
+
+ /**
+ * Creates a lease on the specified resource with given timeout.
+ *
+ * @param resource
+ * Resource for which the lease has to be created
+ * @param timeout
+ * Lease lifetime in milliseconds
+ */
+ public Lease(T resource, long timeout) {
+ this.resource = resource;
+ this.leaseTimeout = timeout;
+ this.callbacks = Collections.synchronizedList(new ArrayList<>());
+ this.creationTime = Time.monotonicNow();
+ this.expired = false;
+ }
+
+ /**
+ * Returns true if the lease has expired, else false.
+ *
+ * @return true if expired, else false
+ */
+ public boolean hasExpired() {
+ return expired;
+ }
+
+ /**
+ * Registers a callback which will be executed in case of timeout. Callbacks
+ * are executed in a separate Thread.
+ *
+ * @param callback
+ * The Callable which has to be executed
+ * @throws LeaseExpiredException
+ * If the lease has already timed out
+ */
+ public void registerCallBack(Callable<Void> callback)
+ throws LeaseExpiredException {
+ if(hasExpired()) {
+ throw new LeaseExpiredException("Resource: " + resource);
+ }
+ callbacks.add(callback);
+ }
+
+ /**
+ * Returns the time elapsed since the creation of lease.
+ *
+ * @return elapsed time in milliseconds
+ * @throws LeaseExpiredException
+ * If the lease has already timed out
+ */
+ public long getElapsedTime() throws LeaseExpiredException {
+ if(hasExpired()) {
+ throw new LeaseExpiredException("Resource: " + resource);
+ }
+ return Time.monotonicNow() - creationTime;
+ }
+
+ /**
+ * Returns the time available before timeout.
+ *
+ * @return remaining time in milliseconds
+ * @throws LeaseExpiredException
+ * If the lease has already timed out
+ */
+ public long getRemainingTime() throws LeaseExpiredException {
+ if(hasExpired()) {
+ throw new LeaseExpiredException("Resource: " + resource);
+ }
+ return leaseTimeout - getElapsedTime();
+ }
+
+ /**
+ * Returns total lease lifetime.
+ *
+ * @return total lifetime of lease in milliseconds
+ * @throws LeaseExpiredException
+ * If the lease has already timed out
+ */
+ public long getLeaseLifeTime() throws LeaseExpiredException {
+ if(hasExpired()) {
+ throw new LeaseExpiredException("Resource: " + resource);
+ }
+ return leaseTimeout;
+ }
+
+ /**
+ * Renews the lease timeout period.
+ *
+ * @param timeout
+ * Time to be added to the lease in milliseconds
+ * @throws LeaseExpiredException
+ * If the lease has already timed out
+ */
+ public void renew(long timeout) throws LeaseExpiredException {
+ if(hasExpired()) {
+ throw new LeaseExpiredException("Resource: " + resource);
+ }
+ leaseTimeout += timeout;
+ }
+
+ @Override
+ public int hashCode() {
+ return resource.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if(obj instanceof Lease) {
+ return resource.equals(((Lease) obj).resource);
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return "Lease<" + resource.toString() + ">";
+ }
+
+ /**
+ * Returns the callbacks to be executed for the lease in case of timeout.
+ *
+ * @return callbacks to be executed
+ */
+ List<Callable<Void>> getCallbacks() {
+ return callbacks;
+ }
+
+ /**
+ * Expires/Invalidates the lease.
+ */
+ void invalidate() {
+ callbacks = null;
+ expired = true;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseAlreadyExistException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseAlreadyExistException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseAlreadyExistException.java
new file mode 100644
index 0000000..a39ea22
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseAlreadyExistException.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <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.lease;
+
+/**
+ * This exception represents that there is already a lease acquired on the
+ * same resource.
+ */
+public class LeaseAlreadyExistException extends LeaseException {
+
+ /**
+ * Constructs an {@code LeaseAlreadyExistException} with {@code null}
+ * as its error detail message.
+ */
+ public LeaseAlreadyExistException() {
+ super();
+ }
+
+ /**
+ * Constructs an {@code LeaseAlreadyExistException} with the specified
+ * detail message.
+ *
+ * @param message
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public LeaseAlreadyExistException(String message) {
+ super(message);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseCallbackExecutor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseCallbackExecutor.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseCallbackExecutor.java
new file mode 100644
index 0000000..1b7391b
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseCallbackExecutor.java
@@ -0,0 +1,65 @@
+/**
+ * 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.lease;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+
+/**
+ * This class is responsible for executing the callbacks of a lease in case of
+ * timeout.
+ */
+public class LeaseCallbackExecutor<T> implements Runnable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Lease.class);
+
+ private final T resource;
+ private final List<Callable<Void>> callbacks;
+
+ /**
+ * Constructs LeaseCallbackExecutor instance with list of callbacks.
+ *
+ * @param resource
+ * The resource for which the callbacks are executed
+ * @param callbacks
+ * Callbacks to be executed by this executor
+ */
+ public LeaseCallbackExecutor(T resource, List<Callable<Void>> callbacks) {
+ this.resource = resource;
+ this.callbacks = callbacks;
+ }
+
+ @Override
+ public void run() {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Executing callbacks for lease on {}", resource);
+ }
+ for(Callable<Void> callback : callbacks) {
+ try {
+ callback.call();
+ } catch (Exception e) {
+ LOG.warn("Exception while executing callback for lease on {}",
+ resource, e);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseException.java
new file mode 100644
index 0000000..418f412
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseException.java
@@ -0,0 +1,45 @@
+/**
+ * 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.lease;
+
+/**
+ * This exception represents all lease related exceptions.
+ */
+public class LeaseException extends Exception {
+
+ /**
+ * Constructs an {@code LeaseException} with {@code null}
+ * as its error detail message.
+ */
+ public LeaseException() {
+ super();
+ }
+
+ /**
+ * Constructs an {@code LeaseException} with the specified
+ * detail message.
+ *
+ * @param message
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public LeaseException(String message) {
+ super(message);
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseExpiredException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseExpiredException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseExpiredException.java
new file mode 100644
index 0000000..440a023
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseExpiredException.java
@@ -0,0 +1,45 @@
+/**
+ * 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.lease;
+
+/**
+ * This exception represents that the lease that is being accessed has expired.
+ */
+public class LeaseExpiredException extends LeaseException {
+
+ /**
+ * Constructs an {@code LeaseExpiredException} with {@code null}
+ * as its error detail message.
+ */
+ public LeaseExpiredException() {
+ super();
+ }
+
+ /**
+ * Constructs an {@code LeaseExpiredException} with the specified
+ * detail message.
+ *
+ * @param message
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public LeaseExpiredException(String message) {
+ super(message);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java
new file mode 100644
index 0000000..b8390dd
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java
@@ -0,0 +1,247 @@
+/**
+ * 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.lease;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * LeaseManager is someone who can provide you leases based on your
+ * requirement. If you want to return the lease back before it expires,
+ * you can give it back to Lease Manager. He is the one responsible for
+ * the lifecycle of leases. The resource for which lease is created
+ * should have proper {@code equals} method implementation, resource
+ * equality is checked while the lease is created.
+ *
+ * @param <T> Type of leases that this lease manager can create
+ */
+public class LeaseManager<T> {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(LeaseManager.class);
+
+ private final long defaultTimeout;
+ private Map<T, Lease<T>> activeLeases;
+ private LeaseMonitor leaseMonitor;
+ private Thread leaseMonitorThread;
+ private boolean isRunning;
+
+ /**
+ * Creates an instance of lease manager.
+ *
+ * @param defaultTimeout
+ * Default timeout in milliseconds to be used for lease creation.
+ */
+ public LeaseManager(long defaultTimeout) {
+ this.defaultTimeout = defaultTimeout;
+ }
+
+ /**
+ * Starts the lease manager service.
+ */
+ public void start() {
+ LOG.debug("Starting LeaseManager service");
+ activeLeases = new ConcurrentHashMap<>();
+ leaseMonitor = new LeaseMonitor();
+ leaseMonitorThread = new Thread(leaseMonitor);
+ leaseMonitorThread.setName("LeaseManager#LeaseMonitor");
+ leaseMonitorThread.setDaemon(true);
+ leaseMonitorThread.setUncaughtExceptionHandler((thread, throwable) -> {
+ // Let us just restart this thread after logging an error.
+ // if this thread is not running we cannot handle Lease expiry.
+ LOG.error("LeaseMonitor thread encountered an error. Thread: {}",
+ thread.toString(), throwable);
+ leaseMonitorThread.start();
+ });
+ LOG.debug("Starting LeaseManager#LeaseMonitor Thread");
+ leaseMonitorThread.start();
+ isRunning = true;
+ }
+
+ /**
+ * Returns a lease for the specified resource with default timeout.
+ *
+ * @param resource
+ * Resource for which lease has to be created
+ * @throws LeaseAlreadyExistException
+ * If there is already a lease on the resource
+ */
+ public synchronized Lease<T> acquire(T resource)
+ throws LeaseAlreadyExistException {
+ return acquire(resource, defaultTimeout);
+ }
+
+ /**
+ * Returns a lease for the specified resource with the timeout provided.
+ *
+ * @param resource
+ * Resource for which lease has to be created
+ * @param timeout
+ * The timeout in milliseconds which has to be set on the lease
+ * @throws LeaseAlreadyExistException
+ * If there is already a lease on the resource
+ */
+ public synchronized Lease<T> acquire(T resource, long timeout)
+ throws LeaseAlreadyExistException {
+ checkStatus();
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Acquiring lease on {} for {} milliseconds", resource, timeout);
+ }
+ if(activeLeases.containsKey(resource)) {
+ throw new LeaseAlreadyExistException("Resource: " + resource);
+ }
+ Lease<T> lease = new Lease<>(resource, timeout);
+ activeLeases.put(resource, lease);
+ leaseMonitorThread.interrupt();
+ return lease;
+ }
+
+ /**
+ * Returns a lease associated with the specified resource.
+ *
+ * @param resource
+ * Resource for which the lease has to be returned
+ * @throws LeaseNotFoundException
+ * If there is no active lease on the resource
+ */
+ public Lease<T> get(T resource) throws LeaseNotFoundException {
+ checkStatus();
+ Lease<T> lease = activeLeases.get(resource);
+ if(lease != null) {
+ return lease;
+ }
+ throw new LeaseNotFoundException("Resource: " + resource);
+ }
+
+ /**
+ * Releases the lease associated with the specified resource.
+ *
+ * @param resource
+ * The for which the lease has to be released
+ * @throws LeaseNotFoundException
+ * If there is no active lease on the resource
+ */
+ public synchronized void release(T resource)
+ throws LeaseNotFoundException {
+ checkStatus();
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Releasing lease on {}", resource);
+ }
+ Lease<T> lease = activeLeases.remove(resource);
+ if(lease == null) {
+ throw new LeaseNotFoundException("Resource: " + resource);
+ }
+ lease.invalidate();
+ }
+
+ /**
+ * Shuts down the LeaseManager and releases the resources. All the active
+ * {@link Lease} will be released (callbacks on leases will not be
+ * executed).
+ */
+ public void shutdown() {
+ checkStatus();
+ LOG.debug("Shutting down LeaseManager service");
+ leaseMonitor.disable();
+ leaseMonitorThread.interrupt();
+ for(T resource : activeLeases.keySet()) {
+ try {
+ release(resource);
+ } catch(LeaseNotFoundException ex) {
+ //Ignore the exception, someone might have released the lease
+ }
+ }
+ isRunning = false;
+ }
+
+ /**
+ * Throws {@link LeaseManagerNotRunningException} if the service is not
+ * running.
+ */
+ private void checkStatus() {
+ if(!isRunning) {
+ throw new LeaseManagerNotRunningException("LeaseManager not running.");
+ }
+ }
+
+ /**
+ * Monitors the leases and expires them based on the timeout, also
+ * responsible for executing the callbacks of expired leases.
+ */
+ private final class LeaseMonitor implements Runnable {
+
+ private boolean monitor = true;
+ private ExecutorService executorService;
+
+ private LeaseMonitor() {
+ this.monitor = true;
+ this.executorService = Executors.newCachedThreadPool();
+ }
+
+ @Override
+ public void run() {
+ while(monitor) {
+ LOG.debug("LeaseMonitor: checking for lease expiry");
+ long sleepTime = Long.MAX_VALUE;
+
+ for (T resource : activeLeases.keySet()) {
+ try {
+ Lease<T> lease = get(resource);
+ long remainingTime = lease.getRemainingTime();
+ if (remainingTime <= 0) {
+ //Lease has timed out
+ List<Callable<Void>> leaseCallbacks = lease.getCallbacks();
+ release(resource);
+ executorService.execute(
+ new LeaseCallbackExecutor(resource, leaseCallbacks));
+ } else {
+ sleepTime = remainingTime > sleepTime ?
+ sleepTime : remainingTime;
+ }
+ } catch (LeaseNotFoundException | LeaseExpiredException ex) {
+ //Ignore the exception, someone might have released the lease
+ }
+ }
+
+ try {
+ if(!Thread.interrupted()) {
+ Thread.sleep(sleepTime);
+ }
+ } catch (InterruptedException ignored) {
+ // This means a new lease is added to activeLeases.
+ }
+ }
+ }
+
+ /**
+ * Disables lease monitor, next interrupt call on the thread
+ * will stop lease monitor.
+ */
+ public void disable() {
+ monitor = false;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManagerNotRunningException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManagerNotRunningException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManagerNotRunningException.java
new file mode 100644
index 0000000..ced31de
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManagerNotRunningException.java
@@ -0,0 +1,45 @@
+/**
+ * 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.lease;
+
+/**
+ * This exception represents that there LeaseManager service is not running.
+ */
+public class LeaseManagerNotRunningException extends RuntimeException {
+
+ /**
+ * Constructs an {@code LeaseManagerNotRunningException} with {@code null}
+ * as its error detail message.
+ */
+ public LeaseManagerNotRunningException() {
+ super();
+ }
+
+ /**
+ * Constructs an {@code LeaseManagerNotRunningException} with the specified
+ * detail message.
+ *
+ * @param message
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public LeaseManagerNotRunningException(String message) {
+ super(message);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseNotFoundException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseNotFoundException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseNotFoundException.java
new file mode 100644
index 0000000..c292d33
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseNotFoundException.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <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.lease;
+
+/**
+ * This exception represents that the lease that is being accessed does not
+ * exist.
+ */
+public class LeaseNotFoundException extends LeaseException {
+
+ /**
+ * Constructs an {@code LeaseNotFoundException} with {@code null}
+ * as its error detail message.
+ */
+ public LeaseNotFoundException() {
+ super();
+ }
+
+ /**
+ * Constructs an {@code LeaseNotFoundException} with the specified
+ * detail message.
+ *
+ * @param message
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public LeaseNotFoundException(String message) {
+ super(message);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/package-info.java
new file mode 100644
index 0000000..48ee2e1
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/package-info.java
@@ -0,0 +1,26 @@
+/**
+ * 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.
+ */
+
+/**
+ * A generic lease management API which can be used if a service
+ * needs any kind of lease management.
+ */
+
+package org.apache.hadoop.ozone.lease;
+/*
+ This package contains lease management related classes.
+ */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/package-info.java
new file mode 100644
index 0000000..db399db
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/package-info.java
@@ -0,0 +1,35 @@
+/**
+ * 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;
+
+/**
+ This package contains class that support ozone implementation on the datanode
+ side.
+
+ Main parts of ozone on datanode are:
+
+ 1. REST Interface - This code lives under the web directory and listens to the
+ WebHDFS port.
+
+ 2. Datanode container classes: This support persistence of ozone objects on
+ datanode. These classes live under container directory.
+
+ 3. Client and Shell: We also support a ozone REST client lib, they are under
+ web/client and web/ozShell.
+
+ */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
new file mode 100644
index 0000000..fa79341
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
@@ -0,0 +1,170 @@
+/**
+ * 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.protocolPB;
+
+import com.google.common.collect.Sets;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
+import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .AllocateScmBlockRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .AllocateScmBlockResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .DeleteKeyBlocksResultProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .DeleteScmKeyBlocksRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .DeleteScmKeyBlocksResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .GetScmBlockLocationsRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .GetScmBlockLocationsResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos
+ .ScmLocatedBlockProto;
+import org.apache.hadoop.ozone.common.BlockGroup;
+import org.apache.hadoop.ozone.common.DeleteBlockGroupResult;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is the server-side translator that forwards requests received on
+ * {@link StorageContainerLocationProtocolPB} to the
+ * {@link StorageContainerLocationProtocol} server implementation.
+ */
+@InterfaceAudience.Private
+public final class ScmBlockLocationProtocolServerSideTranslatorPB
+ implements ScmBlockLocationProtocolPB {
+
+ private final ScmBlockLocationProtocol impl;
+
+ /**
+ * Creates a new ScmBlockLocationProtocolServerSideTranslatorPB.
+ *
+ * @param impl {@link ScmBlockLocationProtocol} server implementation
+ */
+ public ScmBlockLocationProtocolServerSideTranslatorPB(
+ ScmBlockLocationProtocol impl) throws IOException {
+ this.impl = impl;
+ }
+
+
+ @Override
+ public GetScmBlockLocationsResponseProto getScmBlockLocations(
+ RpcController controller, GetScmBlockLocationsRequestProto req)
+ throws ServiceException {
+ Set<String> keys = Sets.newLinkedHashSetWithExpectedSize(
+ req.getKeysCount());
+ for (String key : req.getKeysList()) {
+ keys.add(key);
+ }
+ final Set<AllocatedBlock> blocks;
+ try {
+ blocks = impl.getBlockLocations(keys);
+ } catch (IOException ex) {
+ throw new ServiceException(ex);
+ }
+ GetScmBlockLocationsResponseProto.Builder resp =
+ GetScmBlockLocationsResponseProto.newBuilder();
+ for (AllocatedBlock block: blocks) {
+ ScmLocatedBlockProto.Builder locatedBlock =
+ ScmLocatedBlockProto.newBuilder()
+ .setKey(block.getKey())
+ .setPipeline(block.getPipeline().getProtobufMessage());
+ resp.addLocatedBlocks(locatedBlock.build());
+ }
+ return resp.build();
+ }
+
+ @Override
+ public AllocateScmBlockResponseProto allocateScmBlock(
+ RpcController controller, AllocateScmBlockRequestProto request)
+ throws ServiceException {
+ try {
+ AllocatedBlock allocatedBlock =
+ impl.allocateBlock(request.getSize(), request.getType(),
+ request.getFactor(), request.getOwner());
+ if (allocatedBlock != null) {
+ return
+ AllocateScmBlockResponseProto.newBuilder()
+ .setKey(allocatedBlock.getKey())
+ .setPipeline(allocatedBlock.getPipeline().getProtobufMessage())
+ .setCreateContainer(allocatedBlock.getCreateContainer())
+ .setErrorCode(AllocateScmBlockResponseProto.Error.success)
+ .build();
+ } else {
+ return AllocateScmBlockResponseProto.newBuilder()
+ .setErrorCode(AllocateScmBlockResponseProto.Error.unknownFailure)
+ .build();
+ }
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public DeleteScmKeyBlocksResponseProto deleteScmKeyBlocks(
+ RpcController controller, DeleteScmKeyBlocksRequestProto req)
+ throws ServiceException {
+ DeleteScmKeyBlocksResponseProto.Builder resp =
+ DeleteScmKeyBlocksResponseProto.newBuilder();
+ try {
+ List<BlockGroup> infoList = req.getKeyBlocksList().stream()
+ .map(BlockGroup::getFromProto).collect(Collectors.toList());
+ final List<DeleteBlockGroupResult> results =
+ impl.deleteKeyBlocks(infoList);
+ for (DeleteBlockGroupResult result: results) {
+ DeleteKeyBlocksResultProto.Builder deleteResult =
+ DeleteKeyBlocksResultProto
+ .newBuilder()
+ .setObjectKey(result.getObjectKey())
+ .addAllBlockResults(result.getBlockResultProtoList());
+ resp.addResults(deleteResult.build());
+ }
+ } catch (IOException ex) {
+ throw new ServiceException(ex);
+ }
+ return resp.build();
+ }
+
+ @Override
+ public HddsProtos.GetScmInfoRespsonseProto getScmInfo(
+ RpcController controller, HddsProtos.GetScmInfoRequestProto req)
+ throws ServiceException {
+ ScmInfo scmInfo;
+ try {
+ scmInfo = impl.getScmInfo();
+ } catch (IOException ex) {
+ throw new ServiceException(ex);
+ }
+ return HddsProtos.GetScmInfoRespsonseProto.newBuilder()
+ .setClusterId(scmInfo.getClusterId())
+ .setScmId(scmInfo.getScmId())
+ .build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
new file mode 100644
index 0000000..4974268
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
@@ -0,0 +1,212 @@
+
+/**
+ * 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.protocolPB;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
+import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.ContainerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.ContainerResponseProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.GetContainerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.GetContainerResponseProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.ObjectStageChangeRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.ObjectStageChangeResponseProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.PipelineRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.PipelineResponseProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.SCMDeleteContainerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.SCMDeleteContainerResponseProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.SCMListContainerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerLocationProtocolProtos.SCMListContainerResponseProto;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.List;
+
+/**
+ * This class is the server-side translator that forwards requests received on
+ * {@link StorageContainerLocationProtocolPB} to the
+ * {@link StorageContainerLocationProtocol} server implementation.
+ */
+@InterfaceAudience.Private
+public final class StorageContainerLocationProtocolServerSideTranslatorPB
+ implements StorageContainerLocationProtocolPB {
+
+ private final StorageContainerLocationProtocol impl;
+
+ /**
+ * Creates a new StorageContainerLocationProtocolServerSideTranslatorPB.
+ *
+ * @param impl {@link StorageContainerLocationProtocol} server implementation
+ */
+ public StorageContainerLocationProtocolServerSideTranslatorPB(
+ StorageContainerLocationProtocol impl) throws IOException {
+ this.impl = impl;
+ }
+
+ @Override
+ public ContainerResponseProto allocateContainer(RpcController unused,
+ ContainerRequestProto request) throws ServiceException {
+ try {
+ Pipeline pipeline = impl.allocateContainer(request.getReplicationType(),
+ request.getReplicationFactor(), request.getContainerName(),
+ request.getOwner());
+ return ContainerResponseProto.newBuilder()
+ .setPipeline(pipeline.getProtobufMessage())
+ .setErrorCode(ContainerResponseProto.Error.success)
+ .build();
+
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public GetContainerResponseProto getContainer(
+ RpcController controller, GetContainerRequestProto request)
+ throws ServiceException {
+ try {
+ Pipeline pipeline = impl.getContainer(request.getContainerName());
+ return GetContainerResponseProto.newBuilder()
+ .setPipeline(pipeline.getProtobufMessage())
+ .build();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public SCMListContainerResponseProto listContainer(RpcController controller,
+ SCMListContainerRequestProto request) throws ServiceException {
+ try {
+ String startName = null;
+ String prefixName = null;
+ int count = -1;
+
+ // Arguments check.
+ if (request.hasPrefixName()) {
+ // End container name is given.
+ prefixName = request.getPrefixName();
+ }
+ if (request.hasStartName()) {
+ // End container name is given.
+ startName = request.getStartName();
+ }
+
+ count = request.getCount();
+ List<ContainerInfo> containerList =
+ impl.listContainer(startName, prefixName, count);
+ SCMListContainerResponseProto.Builder builder =
+ SCMListContainerResponseProto.newBuilder();
+ for (ContainerInfo container : containerList) {
+ builder.addContainers(container.getProtobuf());
+ }
+ return builder.build();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public SCMDeleteContainerResponseProto deleteContainer(
+ RpcController controller, SCMDeleteContainerRequestProto request)
+ throws ServiceException {
+ try {
+ impl.deleteContainer(request.getContainerName());
+ return SCMDeleteContainerResponseProto.newBuilder().build();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public StorageContainerLocationProtocolProtos.NodeQueryResponseProto
+ queryNode(RpcController controller,
+ StorageContainerLocationProtocolProtos.NodeQueryRequestProto request)
+ throws ServiceException {
+ try {
+ EnumSet<HddsProtos.NodeState> nodeStateEnumSet = EnumSet.copyOf(request
+ .getQueryList());
+ HddsProtos.NodePool datanodes = impl.queryNode(nodeStateEnumSet,
+ request.getScope(), request.getPoolName());
+ return StorageContainerLocationProtocolProtos
+ .NodeQueryResponseProto.newBuilder()
+ .setDatanodes(datanodes)
+ .build();
+ } catch (Exception e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public ObjectStageChangeResponseProto notifyObjectStageChange(
+ RpcController controller, ObjectStageChangeRequestProto request)
+ throws ServiceException {
+ try {
+ impl.notifyObjectStageChange(request.getType(), request.getName(),
+ request.getOp(), request.getStage());
+ return ObjectStageChangeResponseProto.newBuilder().build();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public PipelineResponseProto allocatePipeline(
+ RpcController controller, PipelineRequestProto request)
+ throws ServiceException {
+ // TODO : Wiring this up requires one more patch.
+ return null;
+ }
+
+ @Override
+ public HddsProtos.GetScmInfoRespsonseProto getScmInfo(
+ RpcController controller, HddsProtos.GetScmInfoRequestProto req)
+ throws ServiceException {
+ try {
+ ScmInfo scmInfo = impl.getScmInfo();
+ return HddsProtos.GetScmInfoRespsonseProto.newBuilder()
+ .setClusterId(scmInfo.getClusterId())
+ .setScmId(scmInfo.getScmId())
+ .build();
+ } catch (IOException ex) {
+ throw new ServiceException(ex);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/package-info.java
new file mode 100644
index 0000000..860386d
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/protocolPB/package-info.java
@@ -0,0 +1,24 @@
+/**
+ * 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.protocolPB;
+
+/**
+ * This package contains classes for the Protocol Buffers binding of Ozone
+ * protocols.
+ */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/web/utils/JsonUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/web/utils/JsonUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/web/utils/JsonUtils.java
new file mode 100644
index 0000000..af56da3
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/web/utils/JsonUtils.java
@@ -0,0 +1,71 @@
+/**
+ * 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.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.type.CollectionType;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * JSON Utility functions used in ozone.
+ */
+public final class JsonUtils {
+
+ // Reuse ObjectMapper instance for improving performance.
+ // ObjectMapper is thread safe as long as we always configure instance
+ // before use.
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final ObjectReader READER = MAPPER.readerFor(Object.class);
+ private static final ObjectWriter WRITTER =
+ MAPPER.writerWithDefaultPrettyPrinter();
+
+ private JsonUtils() {
+ // Never constructed
+ }
+
+ public static String toJsonStringWithDefaultPrettyPrinter(String jsonString)
+ throws IOException {
+ Object json = READER.readValue(jsonString);
+ return WRITTER.writeValueAsString(json);
+ }
+
+ public static String toJsonString(Object obj) throws IOException {
+ return MAPPER.writeValueAsString(obj);
+ }
+
+ /**
+ * Deserialize a list of elements from a given string,
+ * each element in the list is in the given type.
+ *
+ * @param str json string.
+ * @param elementType element type.
+ * @return List of elements of type elementType
+ * @throws IOException
+ */
+ public static List<?> toJsonList(String str, Class<?> elementType)
+ throws IOException {
+ CollectionType type = MAPPER.getTypeFactory()
+ .constructCollectionType(List.class, elementType);
+ return MAPPER.readValue(str, type);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundService.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundService.java
new file mode 100644
index 0000000..431da64
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundService.java
@@ -0,0 +1,161 @@
+/**
+ * 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.utils;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * An abstract class for a background service in ozone.
+ * A background service schedules multiple child tasks in parallel
+ * in a certain period. In each interval, it waits until all the tasks
+ * finish execution and then schedule next interval.
+ */
+public abstract class BackgroundService {
+
+ @VisibleForTesting
+ public static final Logger LOG =
+ LoggerFactory.getLogger(BackgroundService.class);
+
+ // Executor to launch child tasks
+ private final ScheduledExecutorService exec;
+ private final ThreadGroup threadGroup;
+ private final ThreadFactory threadFactory;
+ private final String serviceName;
+ private final long interval;
+ private final long serviceTimeout;
+ private final TimeUnit unit;
+ private final PeriodicalTask service;
+
+ public BackgroundService(String serviceName, long interval,
+ TimeUnit unit, int threadPoolSize, long serviceTimeout) {
+ this.interval = interval;
+ this.unit = unit;
+ this.serviceName = serviceName;
+ this.serviceTimeout = serviceTimeout;
+ threadGroup = new ThreadGroup(serviceName);
+ ThreadFactory tf = r -> new Thread(threadGroup, r);
+ threadFactory = new ThreadFactoryBuilder()
+ .setThreadFactory(tf)
+ .setDaemon(true)
+ .setNameFormat(serviceName + "#%d")
+ .build();
+ exec = Executors.newScheduledThreadPool(threadPoolSize, threadFactory);
+ service = new PeriodicalTask();
+ }
+
+ protected ExecutorService getExecutorService() {
+ return this.exec;
+ }
+
+ @VisibleForTesting
+ public int getThreadCount() {
+ return threadGroup.activeCount();
+ }
+
+ @VisibleForTesting
+ public void triggerBackgroundTaskForTesting() {
+ service.run();
+ }
+
+ // start service
+ public void start() {
+ exec.scheduleWithFixedDelay(service, 0, interval, unit);
+ }
+
+ public abstract BackgroundTaskQueue getTasks();
+
+ /**
+ * Run one or more background tasks concurrently.
+ * Wait until all tasks to return the result.
+ */
+ public class PeriodicalTask implements Runnable {
+ @Override
+ public synchronized void run() {
+ LOG.debug("Running background service : {}", serviceName);
+ BackgroundTaskQueue tasks = getTasks();
+ if (tasks.isEmpty()) {
+ // No task found, or some problems to init tasks
+ // return and retry in next interval.
+ return;
+ }
+
+ LOG.debug("Number of background tasks to execute : {}", tasks.size());
+ CompletionService<BackgroundTaskResult> taskCompletionService =
+ new ExecutorCompletionService<>(exec);
+
+ List<Future<BackgroundTaskResult>> results = Lists.newArrayList();
+ while (tasks.size() > 0) {
+ BackgroundTask task = tasks.poll();
+ Future<BackgroundTaskResult> result =
+ taskCompletionService.submit(task);
+ results.add(result);
+ }
+
+ results.parallelStream().forEach(taskResultFuture -> {
+ try {
+ // Collect task results
+ BackgroundTaskResult result = serviceTimeout > 0
+ ? taskResultFuture.get(serviceTimeout, TimeUnit.MILLISECONDS)
+ : taskResultFuture.get();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("task execution result size {}", result.getSize());
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(
+ "Background task fails to execute, "
+ + "retrying in next interval", e);
+ } catch (TimeoutException e) {
+ LOG.warn("Background task executes timed out, "
+ + "retrying in next interval", e);
+ }
+ });
+ }
+ }
+
+ // shutdown and make sure all threads are properly released.
+ public void shutdown() {
+ LOG.info("Shutting down service {}", this.serviceName);
+ exec.shutdown();
+ try {
+ if (!exec.awaitTermination(60, TimeUnit.SECONDS)) {
+ exec.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ exec.shutdownNow();
+ }
+ if (threadGroup.activeCount() == 0 && !threadGroup.isDestroyed()) {
+ threadGroup.destroy();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTask.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTask.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTask.java
new file mode 100644
index 0000000..47e8ebc
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTask.java
@@ -0,0 +1,28 @@
+/**
+ * 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.utils;
+
+import java.util.concurrent.Callable;
+
+/**
+ * A task thread to run by {@link BackgroundService}.
+ */
+public interface BackgroundTask<T> extends Callable<T> {
+
+ int getPriority();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskQueue.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskQueue.java
new file mode 100644
index 0000000..b56ef0c
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskQueue.java
@@ -0,0 +1,64 @@
+/**
+ * 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.utils;
+
+import java.util.PriorityQueue;
+
+/**
+ * A priority queue that stores a number of {@link BackgroundTask}.
+ */
+public class BackgroundTaskQueue {
+
+ private final PriorityQueue<BackgroundTask> tasks;
+
+ public BackgroundTaskQueue() {
+ tasks = new PriorityQueue<>((task1, task2)
+ -> task1.getPriority() - task2.getPriority());
+ }
+
+ /**
+ * @return the head task in this queue.
+ */
+ public synchronized BackgroundTask poll() {
+ return tasks.poll();
+ }
+
+ /**
+ * Add a {@link BackgroundTask} to the queue,
+ * the task will be sorted by its priority.
+ *
+ * @param task
+ */
+ public synchronized void add(BackgroundTask task) {
+ tasks.add(task);
+ }
+
+ /**
+ * @return true if the queue contains no task, false otherwise.
+ */
+ public synchronized boolean isEmpty() {
+ return tasks.isEmpty();
+ }
+
+ /**
+ * @return the size of the queue.
+ */
+ public synchronized int size() {
+ return tasks.size();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskResult.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskResult.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskResult.java
new file mode 100644
index 0000000..198300f
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BackgroundTaskResult.java
@@ -0,0 +1,44 @@
+/**
+ * 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.utils;
+
+/**
+ * Result of a {@link BackgroundTask}.
+ */
+public interface BackgroundTaskResult {
+
+ /**
+ * Returns the size of entries included in this result.
+ */
+ int getSize();
+
+ /**
+ * An empty task result implementation.
+ */
+ class EmptyTaskResult implements BackgroundTaskResult {
+
+ public static EmptyTaskResult newResult() {
+ return new EmptyTaskResult();
+ }
+
+ @Override
+ public int getSize() {
+ return 0;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BatchOperation.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BatchOperation.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BatchOperation.java
new file mode 100644
index 0000000..47699eb
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/BatchOperation.java
@@ -0,0 +1,90 @@
+/*
+ * 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.utils;
+
+import com.google.common.collect.Lists;
+
+import java.util.List;
+
+/**
+ * An utility class to store a batch of DB write operations.
+ */
+public class BatchOperation {
+
+ /**
+ * Enum for write operations.
+ */
+ public enum Operation {
+ DELETE, PUT
+ }
+
+ private List<SingleOperation> operations =
+ Lists.newArrayList();
+
+ /**
+ * Add a PUT operation into the batch.
+ */
+ public void put(byte[] key, byte[] value) {
+ operations.add(new SingleOperation(Operation.PUT, key, value));
+ }
+
+ /**
+ * Add a DELETE operation into the batch.
+ */
+ public void delete(byte[] key) {
+ operations.add(new SingleOperation(Operation.DELETE, key, null));
+
+ }
+
+ public List<SingleOperation> getOperations() {
+ return operations;
+ }
+
+ /**
+ * A SingleOperation represents a PUT or DELETE operation
+ * and the data the operation needs to manipulates.
+ */
+ public static class SingleOperation {
+
+ private Operation opt;
+ private byte[] key;
+ private byte[] value;
+
+ public SingleOperation(Operation opt, byte[] key, byte[] value) {
+ this.opt = opt;
+ if (key == null) {
+ throw new IllegalArgumentException("key cannot be null");
+ }
+ this.key = key.clone();
+ this.value = value == null ? null : value.clone();
+ }
+
+ public Operation getOpt() {
+ return opt;
+ }
+
+ public byte[] getKey() {
+ return key.clone();
+ }
+
+ public byte[] getValue() {
+ return value == null ? null : value.clone();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/EntryConsumer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/EntryConsumer.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/EntryConsumer.java
new file mode 100644
index 0000000..c407398
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/EntryConsumer.java
@@ -0,0 +1,38 @@
+/*
+ * 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.utils;
+
+import java.io.IOException;
+
+/**
+ * A consumer for metadata store key-value entries.
+ * Used by {@link MetadataStore} class.
+ */
+@FunctionalInterface
+public interface EntryConsumer {
+
+ /**
+ * Consumes a key and value and produces a boolean result.
+ * @param key key
+ * @param value value
+ * @return a boolean value produced by the consumer
+ * @throws IOException
+ */
+ boolean consume(byte[] key, byte[] value) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStore.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStore.java
new file mode 100644
index 0000000..83ca83d
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStore.java
@@ -0,0 +1,380 @@
+/*
+ * 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.utils;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.hadoop.utils.MetadataKeyFilters.MetadataKeyFilter;
+import org.fusesource.leveldbjni.JniDBFactory;
+import org.iq80.leveldb.DB;
+import org.iq80.leveldb.DBIterator;
+import org.iq80.leveldb.Options;
+import org.iq80.leveldb.ReadOptions;
+import org.iq80.leveldb.Snapshot;
+import org.iq80.leveldb.WriteBatch;
+import org.iq80.leveldb.WriteOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * LevelDB interface.
+ */
+public class LevelDBStore implements MetadataStore {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(LevelDBStore.class);
+
+ private DB db;
+ private final File dbFile;
+ private final Options dbOptions;
+ private final WriteOptions writeOptions;
+
+ public LevelDBStore(File dbPath, boolean createIfMissing)
+ throws IOException {
+ dbOptions = new Options();
+ dbOptions.createIfMissing(createIfMissing);
+ this.dbFile = dbPath;
+ this.writeOptions = new WriteOptions().sync(true);
+ openDB(dbPath, dbOptions);
+ }
+
+ /**
+ * Opens a DB file.
+ *
+ * @param dbPath - DB File path
+ * @throws IOException
+ */
+ public LevelDBStore(File dbPath, Options options)
+ throws IOException {
+ dbOptions = options;
+ this.dbFile = dbPath;
+ this.writeOptions = new WriteOptions().sync(true);
+ openDB(dbPath, dbOptions);
+ }
+
+ private void openDB(File dbPath, Options options) throws IOException {
+ dbPath.getParentFile().mkdirs();
+ db = JniDBFactory.factory.open(dbPath, options);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("LevelDB successfully opened");
+ LOG.debug("[Option] cacheSize = " + options.cacheSize());
+ LOG.debug("[Option] createIfMissing = " + options.createIfMissing());
+ LOG.debug("[Option] blockSize = " + options.blockSize());
+ LOG.debug("[Option] compressionType= " + options.compressionType());
+ LOG.debug("[Option] maxOpenFiles= " + options.maxOpenFiles());
+ LOG.debug("[Option] writeBufferSize= "+ options.writeBufferSize());
+ }
+ }
+
+ /**
+ * Puts a Key into file.
+ *
+ * @param key - key
+ * @param value - value
+ */
+ @Override
+ public void put(byte[] key, byte[] value) {
+ db.put(key, value, writeOptions);
+ }
+
+ /**
+ * Get Key.
+ *
+ * @param key key
+ * @return value
+ */
+ @Override
+ public byte[] get(byte[] key) {
+ return db.get(key);
+ }
+
+ /**
+ * Delete Key.
+ *
+ * @param key - Key
+ */
+ @Override
+ public void delete(byte[] key) {
+ db.delete(key);
+ }
+
+ /**
+ * Closes the DB.
+ *
+ * @throws IOException
+ */
+ @Override
+ public void close() throws IOException {
+ if (db != null){
+ db.close();
+ }
+ }
+
+ /**
+ * Returns true if the DB is empty.
+ *
+ * @return boolean
+ * @throws IOException
+ */
+ @Override
+ public boolean isEmpty() throws IOException {
+ try (DBIterator iter = db.iterator()) {
+ iter.seekToFirst();
+ boolean hasNext = !iter.hasNext();
+ return hasNext;
+ }
+ }
+
+ /**
+ * Returns the actual levelDB object.
+ * @return DB handle.
+ */
+ public DB getDB() {
+ return db;
+ }
+
+ /**
+ * Returns an iterator on all the key-value pairs in the DB.
+ * @return an iterator on DB entries.
+ */
+ public DBIterator getIterator() {
+ return db.iterator();
+ }
+
+
+ @Override
+ public void destroy() throws IOException {
+ close();
+ JniDBFactory.factory.destroy(dbFile, dbOptions);
+ }
+
+ @Override
+ public ImmutablePair<byte[], byte[]> peekAround(int offset,
+ byte[] from) throws IOException, IllegalArgumentException {
+ try (DBIterator it = db.iterator()) {
+ if (from == null) {
+ it.seekToFirst();
+ } else {
+ it.seek(from);
+ }
+ if (!it.hasNext()) {
+ return null;
+ }
+ switch (offset) {
+ case 0:
+ Entry<byte[], byte[]> current = it.next();
+ return new ImmutablePair<>(current.getKey(), current.getValue());
+ case 1:
+ if (it.next() != null && it.hasNext()) {
+ Entry<byte[], byte[]> next = it.peekNext();
+ return new ImmutablePair<>(next.getKey(), next.getValue());
+ }
+ break;
+ case -1:
+ if (it.hasPrev()) {
+ Entry<byte[], byte[]> prev = it.peekPrev();
+ return new ImmutablePair<>(prev.getKey(), prev.getValue());
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Position can only be -1, 0 " + "or 1, but found " + offset);
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void iterate(byte[] from, EntryConsumer consumer)
+ throws IOException {
+ try (DBIterator iter = db.iterator()) {
+ if (from != null) {
+ iter.seek(from);
+ } else {
+ iter.seekToFirst();
+ }
+ while (iter.hasNext()) {
+ Entry<byte[], byte[]> current = iter.next();
+ if (!consumer.consume(current.getKey(),
+ current.getValue())) {
+ break;
+ }
+ }
+ }
+ }
+
+ /**
+ * Compacts the DB by removing deleted keys etc.
+ * @throws IOException if there is an error.
+ */
+ @Override
+ public void compactDB() throws IOException {
+ if(db != null) {
+ // From LevelDB docs : begin == null and end == null means the whole DB.
+ db.compactRange(null, null);
+ }
+ }
+
+ @Override
+ public void writeBatch(BatchOperation operation) throws IOException {
+ List<BatchOperation.SingleOperation> operations =
+ operation.getOperations();
+ if (!operations.isEmpty()) {
+ try (WriteBatch writeBatch = db.createWriteBatch()) {
+ for (BatchOperation.SingleOperation opt : operations) {
+ switch (opt.getOpt()) {
+ case DELETE:
+ writeBatch.delete(opt.getKey());
+ break;
+ case PUT:
+ writeBatch.put(opt.getKey(), opt.getValue());
+ break;
+ default:
+ throw new IllegalArgumentException("Invalid operation "
+ + opt.getOpt());
+ }
+ }
+ db.write(writeBatch);
+ }
+ }
+ }
+
+ @Override
+ public List<Map.Entry<byte[], byte[]>> getRangeKVs(byte[] startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return getRangeKVs(startKey, count, false, filters);
+ }
+
+ @Override
+ public List<Map.Entry<byte[], byte[]>> getSequentialRangeKVs(byte[] startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return getRangeKVs(startKey, count, true, filters);
+ }
+
+ /**
+ * Returns a certain range of key value pairs as a list based on a
+ * startKey or count. Further a {@link MetadataKeyFilter} can be added to
+ * filter keys if necessary. To prevent race conditions while listing
+ * entries, this implementation takes a snapshot and lists the entries from
+ * the snapshot. This may, on the other hand, cause the range result slight
+ * different with actual data if data is updating concurrently.
+ * <p>
+ * If the startKey is specified and found in levelDB, this key and the keys
+ * after this key will be included in the result. If the startKey is null
+ * all entries will be included as long as other conditions are satisfied.
+ * If the given startKey doesn't exist, an empty list will be returned.
+ * <p>
+ * The count argument is to limit number of total entries to return,
+ * the value for count must be an integer greater than 0.
+ * <p>
+ * This method allows to specify one or more {@link MetadataKeyFilter}
+ * to filter keys by certain condition. Once given, only the entries
+ * whose key passes all the filters will be included in the result.
+ *
+ * @param startKey a start key.
+ * @param count max number of entries to return.
+ * @param filters customized one or more {@link MetadataKeyFilter}.
+ * @return a list of entries found in the database or an empty list if the
+ * startKey is invalid.
+ * @throws IOException if there are I/O errors.
+ * @throws IllegalArgumentException if count is less than 0.
+ */
+ private List<Entry<byte[], byte[]>> getRangeKVs(byte[] startKey,
+ int count, boolean sequential, MetadataKeyFilter... filters)
+ throws IOException {
+ List<Entry<byte[], byte[]>> result = new ArrayList<>();
+ long start = System.currentTimeMillis();
+ if (count < 0) {
+ throw new IllegalArgumentException(
+ "Invalid count given " + count + ", count must be greater than 0");
+ }
+ Snapshot snapShot = null;
+ DBIterator dbIter = null;
+ try {
+ snapShot = db.getSnapshot();
+ ReadOptions readOptions = new ReadOptions().snapshot(snapShot);
+ dbIter = db.iterator(readOptions);
+ if (startKey == null) {
+ dbIter.seekToFirst();
+ } else {
+ if (db.get(startKey) == null) {
+ // Key not found, return empty list
+ return result;
+ }
+ dbIter.seek(startKey);
+ }
+ while (dbIter.hasNext() && result.size() < count) {
+ byte[] preKey = dbIter.hasPrev() ? dbIter.peekPrev().getKey() : null;
+ byte[] nextKey = dbIter.hasNext() ? dbIter.peekNext().getKey() : null;
+ Entry<byte[], byte[]> current = dbIter.next();
+
+ if (filters == null) {
+ result.add(current);
+ } else {
+ if (Arrays.asList(filters).stream().allMatch(
+ entry -> entry.filterKey(preKey, current.getKey(), nextKey))) {
+ result.add(current);
+ } else {
+ if (result.size() > 0 && sequential) {
+ // if the caller asks for a sequential range of results,
+ // and we met a dis-match, abort iteration from here.
+ // if result is empty, we continue to look for the first match.
+ break;
+ }
+ }
+ }
+ }
+ } finally {
+ if (snapShot != null) {
+ snapShot.close();
+ }
+ if (dbIter != null) {
+ dbIter.close();
+ }
+ if (LOG.isDebugEnabled()) {
+ if (filters != null) {
+ for (MetadataKeyFilters.MetadataKeyFilter filter : filters) {
+ int scanned = filter.getKeysScannedNum();
+ int hinted = filter.getKeysHintedNum();
+ if (scanned > 0 || hinted > 0) {
+ LOG.debug(
+ "getRangeKVs ({}) numOfKeysScanned={}, numOfKeysHinted={}",
+ filter.getClass().getSimpleName(), filter.getKeysScannedNum(),
+ filter.getKeysHintedNum());
+ }
+ }
+ }
+ long end = System.currentTimeMillis();
+ long timeConsumed = end - start;
+ LOG.debug("Time consumed for getRangeKVs() is {}ms,"
+ + " result length is {}.", timeConsumed, result.size());
+ }
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b832f3c/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetadataKeyFilters.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetadataKeyFilters.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetadataKeyFilters.java
new file mode 100644
index 0000000..3ff0a94
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetadataKeyFilters.java
@@ -0,0 +1,118 @@
+/*
+ * 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.utils;
+
+import com.google.common.base.Strings;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.ozone.OzoneConsts;
+
+/**
+ * An utility class to filter levelDB keys.
+ */
+public final class MetadataKeyFilters {
+
+ private static KeyPrefixFilter deletingKeyFilter =
+ new MetadataKeyFilters.KeyPrefixFilter(OzoneConsts.DELETING_KEY_PREFIX);
+
+ private static KeyPrefixFilter normalKeyFilter =
+ new MetadataKeyFilters.KeyPrefixFilter(OzoneConsts.DELETING_KEY_PREFIX,
+ true);
+
+ private MetadataKeyFilters() {
+ }
+
+ public static KeyPrefixFilter getDeletingKeyFilter() {
+ return deletingKeyFilter;
+ }
+
+ public static KeyPrefixFilter getNormalKeyFilter() {
+ return normalKeyFilter;
+ }
+ /**
+ * Interface for levelDB key filters.
+ */
+ public interface MetadataKeyFilter {
+ /**
+ * Filter levelDB key with a certain condition.
+ *
+ * @param preKey previous key.
+ * @param currentKey current key.
+ * @param nextKey next key.
+ * @return true if a certain condition satisfied, return false otherwise.
+ */
+ boolean filterKey(byte[] preKey, byte[] currentKey, byte[] nextKey);
+
+ default int getKeysScannedNum() {
+ return 0;
+ }
+
+ default int getKeysHintedNum() {
+ return 0;
+ }
+ }
+
+ /**
+ * Utility class to filter key by a string prefix. This filter
+ * assumes keys can be parsed to a string.
+ */
+ public static class KeyPrefixFilter implements MetadataKeyFilter {
+
+ private String keyPrefix = null;
+ private int keysScanned = 0;
+ private int keysHinted = 0;
+ private Boolean negative;
+
+ public KeyPrefixFilter(String keyPrefix) {
+ this(keyPrefix, false);
+ }
+
+ public KeyPrefixFilter(String keyPrefix, boolean negative) {
+ this.keyPrefix = keyPrefix;
+ this.negative = negative;
+ }
+
+ @Override
+ public boolean filterKey(byte[] preKey, byte[] currentKey,
+ byte[] nextKey) {
+ keysScanned++;
+ boolean accept = false;
+ if (Strings.isNullOrEmpty(keyPrefix)) {
+ accept = true;
+ } else {
+ if (currentKey != null &&
+ DFSUtil.bytes2String(currentKey).startsWith(keyPrefix)) {
+ keysHinted++;
+ accept = true;
+ } else {
+ accept = false;
+ }
+ }
+ return (negative) ? !accept : accept;
+ }
+
+ @Override
+ public int getKeysScannedNum() {
+ return keysScanned;
+ }
+
+ @Override
+ public int getKeysHintedNum() {
+ return keysHinted;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org