diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/ApplyFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/ApplyFunction.java new file mode 100644 index 0000000000000..ac3d5edb6351a --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/ApplyFunction.java @@ -0,0 +1,89 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * Represents a function that accepts a value of type T and produces a result of type R. + * This interface extends {@link Async} and provides methods to apply the function + * asynchronously using {@link CompletableFuture}. + * + *

ApplyFunction is used to implement the following semantics:

+ * 
+ * {@code
+ *    T res = doAsync(input);
+ *    // Can use ApplyFunction
+ *    R result = thenApply(res);
+ * }
+ *
+ * + * @param the type of the input to the function + * @param the type of the result of the function + */ +@FunctionalInterface +public interface ApplyFunction extends Async{ + + /** + * Applies this function to the given argument. + * + * @param t the function argument + * @return the function result + * @throws IOException if an I/O error occurs + */ + R apply(T t) throws IOException; + + /** + * Applies this function asynchronously to the result of the given {@link CompletableFuture}. + * The function is executed on the same thread as the completion of the given future. + * + * @param in the input future + * @return a new future that holds the result of the function application + */ + default CompletableFuture apply(CompletableFuture in) { + return in.thenApply(t -> { + try { + return ApplyFunction.this.apply(t); + } catch (IOException e) { + throw warpCompletionException(e); + } + }); + } + + /** + * Applies this function asynchronously to the result of the given {@link CompletableFuture}, + * using the specified executor for the asynchronous computation. + * + * @param in the input future + * @param executor the executor to use for the asynchronous computation + * @return a new future that holds the result of the function application + */ + default CompletableFuture apply(CompletableFuture in, Executor executor) { + return in.thenApplyAsync(t -> { + try { + return ApplyFunction.this.apply(t); + } catch (IOException e) { + throw warpCompletionException(e); + } + }, executor); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/Async.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/Async.java new file mode 100644 index 0000000000000..e184bffaeef6c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/Async.java @@ -0,0 +1,115 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.CompletionException; + +/** + * An interface for asynchronous operations, providing utility methods + * and constants related to asynchronous computations. + * + * @param The type of the result of the asynchronous operation + */ +public interface Async { + + /** + * A thread-local variable to store the {@link CompletableFuture} instance for the current thread. + *

+ * Note: After executing an asynchronous method, the thread stores the CompletableFuture + * of the asynchronous method in the thread's local variable + */ + ThreadLocal> CUR_COMPLETABLE_FUTURE + = new ThreadLocal<>(); + + /** + * Sets the {@link CompletableFuture} instance for the current thread. + * + * @param completableFuture The {@link CompletableFuture} instance to be set + * @param The type of the result in the CompletableFuture + */ + default void setCurCompletableFuture(CompletableFuture completableFuture) { + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) completableFuture); + } + + /** + * Gets the {@link CompletableFuture} instance for the current thread. + * + * @return The {@link CompletableFuture} instance for the current thread, + * or {@code null} if not set + */ + default CompletableFuture getCurCompletableFuture() { + return (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + } + + /** + * Blocks and retrieves the result of the {@link CompletableFuture} instance + * for the current thread. + * + * @return The result of the CompletableFuture, or {@code null} if the thread was interrupted + * @throws IOException If the completion exception to the CompletableFuture + * is an IOException or a subclass of it + */ + default R result() throws IOException { + try { + CompletableFuture completableFuture = + (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + return completableFuture.get(); + } catch (InterruptedException e) { + return null; + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof IOException) { + throw (IOException)cause; + } + throw new IOException(e); + } + } + + /** + * Extracts the real cause of an exception wrapped by CompletionException. + * + * @param e The incoming exception, which may be a CompletionException. + * @return Returns the real cause of the original exception, + * or the original exception if there is no cause. + */ + static Throwable unWarpCompletionException(Throwable e) { + if (e instanceof CompletionException) { + if (e.getCause() != null) { + return e.getCause(); + } + } + return e; + } + + /** + * Wraps the incoming exception in a new CompletionException. + * + * @param e The incoming exception, which may be any type of Throwable. + * @return Returns a new CompletionException with the original exception as its cause. + */ + static CompletionException warpCompletionException(Throwable e) { + if (e instanceof CompletionException) { + return (CompletionException) e; + } + return new CompletionException(e); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncApplyFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncApplyFunction.java new file mode 100644 index 0000000000000..b34a6c479cd60 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncApplyFunction.java @@ -0,0 +1,162 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The AsyncApplyFunction interface represents a function that + * asynchronously accepts a value of type T and produces a result + * of type R. This interface extends {@link ApplyFunction} and is + * designed to be used with asynchronous computation frameworks, + * such as Java's {@link java.util.concurrent.CompletableFuture}. + * + *

An implementation of this interface is expected to perform an + * asynchronous operation and return a result, which is typically + * represented as a {@code CompletableFuture}. This allows for + * non-blocking execution of tasks and is particularly useful for + * I/O operations or any operation that may take a significant amount + * of time to complete.

+ * + *

AsyncApplyFunction is used to implement the following semantics:

+ * 
+ * {@code
+ *    T res = doAsync1(input);
+ *    // Can use AsyncApplyFunction
+ *    R result = doAsync2(res);
+ * }
+ *
+ * + * @param the type of the input to the function + * @param the type of the result of the function + * @see ApplyFunction + * @see java.util.concurrent.CompletableFuture + */ +@FunctionalInterface +public interface AsyncApplyFunction extends ApplyFunction { + + /** + * Asynchronously applies this function to the given argument. + * + *

This method is intended to initiate the function application + * without waiting for the result. It is typically used when the + * result of the operation is not required immediately or when the + * operation is part of a larger asynchronous workflow.

+ * + * @param t the function argument + * @throws IOException if an I/O error occurs during the application + * of the function + */ + void applyAsync(T t) throws IOException; + + /** + * Synchronously applies this function to the given argument and + * returns the result. + * + *

This method waits for the asynchronous operation to complete + * and returns its result. It is useful when the result is needed + * immediately and the calling code cannot proceed without it.

+ * + * @param t the function argument + * @return the result of applying the function to the argument + * @throws IOException if an I/O error occurs during the application + * of the function + */ + @Override + default R apply(T t) throws IOException { + applyAsync(t); + return result(); + } + + /** + * Initiates the asynchronous application of this function to the given result. + *

+ * This method calls applyAsync to start the asynchronous operation and then retrieves + * the current thread's CompletableFuture using getCurCompletableFuture. + * It returns this CompletableFuture, which will be completed with the result of the + * asynchronous operation once it is finished. + *

+ * This method is useful for chaining with other asynchronous operations, as it allows the + * current operation to be part of a larger asynchronous workflow. + * + * @param t the function argument + * @return a CompletableFuture that will be completed with the result of the + * asynchronous operation + * @throws IOException if an I/O error occurs during the initiation of the asynchronous operation + */ + default CompletableFuture async(T t) throws IOException { + applyAsync(t); + CompletableFuture completableFuture = getCurCompletableFuture(); + assert completableFuture != null; + return completableFuture; + } + + /** + * Asynchronously applies this function to the result of the given + * CompletableFuture. + * + *

This method chains the function application to the completion + * of the input future. It returns a new CompletableFuture that + * completes with the function's result when the input future + * completes.

+ * + * @param in the input future + * @return a new CompletableFuture that holds the result of the + * function application + */ + @Override + default CompletableFuture apply(CompletableFuture in) { + return in.thenCompose(t -> { + try { + return async(t); + } catch (IOException e) { + throw warpCompletionException(e); + } + }); + } + + /** + * Asynchronously applies this function to the result of the given + * CompletableFuture, using the specified executor for the + * asynchronous computation. + * + *

This method allows for more control over the execution + * context of the asynchronous operation, such as running the + * operation in a separate thread or thread pool.

+ * + * @param in the input future + * @param executor the executor to use for the asynchronous + * computation + * @return a new CompletableFuture that holds the result of the + * function application + */ + @Override + default CompletableFuture apply(CompletableFuture in, Executor executor) { + return in.thenComposeAsync(t -> { + try { + return async(t); + } catch (IOException e) { + throw warpCompletionException(e); + } + }, executor); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncBiFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncBiFunction.java new file mode 100644 index 0000000000000..3e94736e7f104 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncBiFunction.java @@ -0,0 +1,83 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +/** + * The {@code AsyncBiFunction} interface represents a bi-function that + * asynchronously accepts two arguments and produces a result. This interface + * extends the {@link Async} interface and provides a method to apply the + * function asynchronously. + * + *

Implementations of this interface are expected to perform an + * asynchronous operation that takes two parameters of types T and P, and + * returns a result of type R. The asynchronous operation is typically + * represented as a {@link CompletableFuture} of the result type R.

+ * + *

For example, an implementation of this interface might perform an + * asynchronous computation using the two input parameters and return a + * future representing the result of that computation.

+ * + * @param the type of the first argument to the function + * @param

the type of the second argument to the function + * @param the type of the result of the function + * @see Async + */ +@FunctionalInterface +public interface AsyncBiFunction extends Async{ + + /** + * Asynchronously applies this function to the given arguments. + * + *

This method is intended to initiate the function application + * without waiting for the result. It should be used when the + * operation can be performed in the background, and the result + * is not required immediately.

+ * + * @param t the first argument to the function + * @param p the second argument to the function + * @throws IOException if an I/O error occurs during the application of the function + */ + void applyAsync(T t, P p) throws IOException; + + /** + * Initiates the asynchronous application of this function to the given result. + *

+ * This method calls applyAsync to start the asynchronous operation and then retrieves + * the current thread's CompletableFuture using getCurCompletableFuture. + * It returns this CompletableFuture, which will be completed with the result of the + * asynchronous operation once it is finished. + *

+ * This method is useful for chaining with other asynchronous operations, as it allows the + * current operation to be part of a larger asynchronous workflow. + * + * @param t the first argument to the function + * @param p the second argument to the function + * @return a CompletableFuture that will be completed with the result of the + * asynchronous operation + * @throws IOException if an I/O error occurs during the initiation of the asynchronous operation + */ + default CompletableFuture async(T t, P p) throws IOException { + applyAsync(t, p); + CompletableFuture completableFuture = getCurCompletableFuture(); + assert completableFuture != null; + return completableFuture; + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncCatchFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncCatchFunction.java new file mode 100644 index 0000000000000..01fcc44a19f08 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncCatchFunction.java @@ -0,0 +1,174 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.unWarpCompletionException; +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The AsyncCatchFunction interface represents a function that handles exceptions + * occurring within an asynchronous operation. It extends the CatchFunction + * interface and adds the capability to perform asynchronous exception handling. + * + *

This interface is part of the asynchronous utilities provided by the Hadoop + * Distributed File System (HDFS) Federation router. It is used in conjunction + * with other asynchronous interfaces such as AsyncRun to build complex, + * non-blocking operations.

+ * + *

An implementation of this interface should define how to handle a caught + * exception asynchronously. It takes two parameters: the result of the + * asynchronous operation (if any) and the caught exception. The function can then + * initiate an asynchronous process to handle the exception, which may involve + * logging the error, performing a recovery operation, or any other custom logic.

+ * + *

For example, the applyAsync method is intended to be called when an exception + * is caught during an asynchronous operation. It should initiate the asynchronous + * process to handle the exception without blocking the main execution thread.

+ * + *

The default method apply is provided to allow synchronous operation in case + * the asynchronous handling is not required. It simply calls applyAsync and waits + * for the result.

+ * + *

The default method async is provided to allow chaining with other asynchronous + * operations. It calls applyAsync and returns a CompletableFuture that completes + * with the result of the operation.

+ * + *

AsyncCatchFunction is used to implement the following semantics:

+ * 
+ * {@code
+ *    try{
+ *      R res = doAsync1(input);
+ *    } catch(E e) {
+ *      // Can use AsyncCatchFunction
+ *      R result = doAsync2(res, e);
+ *    }
+ * }
+ *
+ * + * @param the type of the result of the asynchronous operation + * @param the type of the exception to catch, extending Throwable + * @see CatchFunction + * @see AsyncRun + */ +@FunctionalInterface +public interface AsyncCatchFunction + extends CatchFunction { + + /** + * Asynchronously applies this function to the given exception. + * + *

This method is intended to be called when an exception is caught + * during an asynchronous operation. It should initiate the asynchronous process + * to handle the exception without blocking the main execution thread. + * The actual handling of the exception, such as logging the error, performing + * a recovery operation, or any other custom logic, should be implemented in this + * method.

+ * + * @param r the result of the asynchronous operation, if any; may be null + * @param e the caught exception + * @throws IOException if an I/O error occurs during the application of the function + */ + void applyAsync(R r, E e) throws IOException; + + /** + * Synchronously applies this function to the given result and exception. + *

+ * This method first calls {@code applyAsync} to initiate the asynchronous handling + * of the exception. Then, it waits for the asynchronous operation to complete + * by calling {@code result}, which retrieves the result of the current + * thread's {@link CompletableFuture}. + *

+ * + * @param r the result of the asynchronous operation, if any; may be null + * @param e the caught exception + * @return the result after applying the function + * @throws IOException if an I/O error occurs during the application of the function + */ + @Override + default R apply(R r, E e) throws IOException { + applyAsync(r, e); + return result(); + } + + /** + * Initiates the asynchronous application of this function to the given result and exception. + *

+ * This method calls applyAsync to start the asynchronous operation and then retrieves + * the current thread's CompletableFuture using getCurCompletableFuture. + * It returns this CompletableFuture, which will be completed with the result of the + * asynchronous operation once it is finished. + *

+ * This method is useful for chaining with other asynchronous operations, as it allows the + * current operation to be part of a larger asynchronous workflow. + * + * @param r the result of the asynchronous operation, if any; may be null + * @param e the caught exception + * @return a CompletableFuture that will be completed with the result of the + * asynchronous operation + * @throws IOException if an I/O error occurs during the initiation of the asynchronous operation + */ + default CompletableFuture async(R r, E e) throws IOException { + applyAsync(r, e); + CompletableFuture completableFuture = getCurCompletableFuture(); + assert completableFuture != null; + return completableFuture; + } + + /** + * Applies the catch function to a {@link CompletableFuture}, handling exceptions of a + * specified type. + *

+ * This method is a default implementation that provides a way to integrate exception + * handling into a chain of asynchronous operations. It takes a {@code CompletableFuture} + * and a class object representing the exception type to catch. The method then completes + * the future with the result of applying this catch function to the input future and the + * specified exception type. + *

+ * If the input future completes exceptionally with an instance of the specified exception + * type, the catch function is applied to the exception. Otherwise, if the future + * completes with a different type of exception or normally, the original result or + * exception is propagated. + * + * @param in the input {@code CompletableFuture} to which the catch function is applied + * @param eClazz the class object representing the exception type to catch + * @return a new {@code CompletableFuture} that completes with the result of applying + * the catch function, or propagates the original exception if it does not match + * the specified type + */ + @Override + default CompletableFuture apply( + CompletableFuture in, Class eClazz) { + return in.handle((r, e) -> { + if (e == null) { + return in; + } + Throwable readException = unWarpCompletionException(e); + if (eClazz.isInstance(readException)) { + try { + return async(r, (E) readException); + } catch (IOException ex) { + throw warpCompletionException(ex); + } + } + throw warpCompletionException(e); + }).thenCompose(result -> result); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncForEachRun.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncForEachRun.java new file mode 100644 index 0000000000000..322242d1c49c0 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncForEachRun.java @@ -0,0 +1,185 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.Iterator; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The AsyncForEachRun class is part of the asynchronous operation utilities + * within the Hadoop Distributed File System (HDFS) Federation router. + * It provides the functionality to perform asynchronous operations on each + * element of an Iterator, applying a given async function. + * + *

This class is designed to work with other asynchronous interfaces and + * utility classes to enable complex asynchronous workflows. It allows for + * non-blocking execution of tasks, which can improve the performance and + * responsiveness of HDFS operations.

+ * + *

The class implements the AsyncRun interface, which means it can be used + * in asynchronous task chains. It maintains an Iterator of elements to + * process, an asyncDoOnce to apply to each element.

+ * + *

The run method initiates the asynchronous operation, and the doOnce + * method recursively applies the asyncDoOnce to each element and handles + * the results. If the shouldBreak flag is set, the operation is completed + * with the current result.

+ * + *

AsyncForEachRun is used to implement the following semantics:

+ * 
+ * {@code
+ * for (I element : elements) {
+ *     R result = asyncDoOnce(element);
+ * }
+ * return result;
+ * }
+ *
+ * + * @param the type of the elements being iterated over + * @param the type of the final result after applying the thenApply function + * @see AsyncRun + * @see AsyncBiFunction + */ +public class AsyncForEachRun implements AsyncRun { + + // Indicates whether the iteration should be broken immediately + // after the next asynchronous operation is completed. + private boolean shouldBreak = false; + // The Iterator over the elements to process asynchronously. + private Iterator iterator; + // The async function to apply to each element from the iterator. + private AsyncBiFunction, I, R> asyncDoOnce; + + /** + * Initiates the asynchronous foreach operation by starting the iteration process + * over the elements provided by the iterator. This method sets up the initial + * call to doOnce(R) with a null result, which begins the recursive + * application of the async function to each element of the iterator. + * + *

This method is an implementation of the {@link AsyncRun} interface's + * {@code run} method, allowing it to be used in a chain of asynchronous + * operations. It is responsible for starting the asynchronous processing and + * handling the completion of the operation through the internal + * {@link CompletableFuture}.

+ * + *

If an exception occurs during the first call to {@code doOnce}, the + * exception is caught and the internal CompletableFuture is completed + * exceptionally with a {@link CompletionException} wrapping the original + * IOException.

+ * + *

After initiating the operation, the method sets the current thread's + * {@link Async} {@link CompletableFuture} by calling + * {@link #setCurCompletableFuture(CompletableFuture)} with the internal result + * CompletableFuture. This allows other parts of the asynchronous workflow to + * chain further operations or handle the final result once the foreach loop + * completes.

+ * + * @see AsyncRun + * @see Async#setCurCompletableFuture(CompletableFuture) + */ + @Override + public void run() { + if (iterator == null || !iterator.hasNext()) { + setCurCompletableFuture(CompletableFuture.completedFuture(null)); + return; + } + CompletableFuture result; + try { + result = doOnce(iterator.next()); + } catch (IOException ioe) { + result = new CompletableFuture<>(); + result.completeExceptionally(warpCompletionException(ioe)); + } + setCurCompletableFuture(result); + } + + /** + * Recursively applies the async function to the next element of the iterator + * and handles the result. This method is called for each iteration of the + * asynchronous foreach loop, applying the async function to each element + * and chaining the results. + * + *

If the iterator has no more elements, the CompletableFuture held by this + * class is completed with the last result. If an exception occurs during + * the application of the async function, it is propagated to the + * CompletableFuture, which completes exceptionally.

+ * + *

This method is designed to be called by the {@link #run()} method and + * handles the iteration logic, including breaking the loop if the + * {@link #shouldBreak} flag is set to true.

+ * + * @param element The current element from the async function application. + * @throws IOException if an I/O error occurs during the application of the async function. + */ + private CompletableFuture doOnce(I element) throws IOException { + CompletableFuture completableFuture = asyncDoOnce.async(AsyncForEachRun.this, element); + return completableFuture.thenCompose(res -> { + if (shouldBreak || !iterator.hasNext()) { + return completableFuture; + } + try { + return doOnce(iterator.next()); + } catch (IOException e) { + throw warpCompletionException(e); + } + }); + } + + /** + * Triggers the termination of the current asynchronous iteration. + * + *

This method is used to break out of the asynchronous for-each loop + * prematurely. It sets a flag that indicates the iteration should be + * terminated at the earliest opportunity. This is particularly useful when + * the processing logic determines that further iteration is unnecessary + * or when a specific condition has been met.

+ * + *

Once this method is called, the next time the loop is about to process + * a new element, it will check the flag and cease operation, allowing the + * application to move on to the next step or complete the task.

+ */ + public void breakNow() { + shouldBreak = true; + } + + /** + * Sets the Iterator for the elements to be processed in the asynchronous operation. + * + * @param forEach The Iterator over the elements. + * @return The current AsyncForEachRun instance for chaining. + */ + public AsyncForEachRun forEach(Iterator forEach) { + this.iterator = forEach; + return this; + } + + /** + * Sets the async function to apply to each element from the iterator. + * + * @param asyncDo The async function. + * @return The current AsyncForEachRun instance for chaining. + */ + public AsyncForEachRun asyncDo(AsyncBiFunction, I, R> asyncDo) { + this.asyncDoOnce = asyncDo; + return this; + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncRun.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncRun.java new file mode 100644 index 0000000000000..03d39f36d7d91 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncRun.java @@ -0,0 +1,74 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +/** + * The AsyncRun interface represents an asynchronous operation that can be + * executed in the context of the Hadoop Distributed File System (HDFS) + * Federation router. Implementations of this interface are responsible for + * performing a task and completing a {@link CompletableFuture} with the + * result of the operation. + * + *

+ * The {@code run} method of this interface is intended to be used in an + * asynchronous workflow, where the operation may involve I/O tasks or + * other long-running processes. By implementing this interface, classes + * can define custom asynchronous behavior that can be chained with other + * asynchronous operations using utility methods provided by the + * {@link AsyncUtil} class.

+ * + *

+ * For example, an implementation of AsyncRun could perform a non-blocking + * read or write operation to HDFS, and upon completion, it could use + * AsyncUtil methods to handle the result or propagate any exceptions that + * occurred during the operation.

+ * + * @param the type of the result produced by the asynchronous operation + * @see AsyncUtil + */ +@FunctionalInterface +public interface AsyncRun extends Async { + + /** + * Executes the asynchronous operation represented by this AsyncRun instance. + * This method is expected to perform the operation and, upon completion, + * complete the current thread's {@link CompletableFuture} with the result. + * + * @throws IOException if an I/O error occurs during the execution of the operation + */ + void run() throws IOException; + + /** + * Provides an asynchronous version of the {@code run} method, which returns a + * {@link CompletableFuture} representing the result of the operation. + * This method is typically used in an asynchronous workflow to initiate the + * operation without waiting for its completion. + * + * @return a CompletableFuture that completes with the result of the operation + * @throws IOException if an I/O error occurs during the initiation of the operation + */ + default CompletableFuture async() throws IOException { + run(); + CompletableFuture completableFuture = getCurCompletableFuture(); + assert completableFuture != null; + return completableFuture; + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncUtil.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncUtil.java new file mode 100644 index 0000000000000..ec2f7360b3092 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncUtil.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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.Collection; +import java.util.Iterator; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.function.Function; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.CUR_COMPLETABLE_FUTURE; +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The AsyncUtil class provides a collection of utility methods to simplify + * the implementation of asynchronous operations using Java's CompletableFuture. + * It encapsulates common patterns such as applying functions, handling exceptions, + * and executing tasks in a non-blocking manner. This class is designed to work + * with Hadoop's asynchronous router operations in HDFS Federation. + * + *

The utility methods support a fluent-style API, allowing for the chaining of + * asynchronous operations. For example, after an asynchronous operation completes, + * a function can be applied to its result, and the process can continue with + * the new result. This is particularly useful for complex workflows that require + * multiple steps, where each step depends on the completion of the previous one.

+ * + *

The class also provides methods to handle exceptions that may occur during a + * synchronous operation. This ensures that error handling is integrated smoothly + * into the workflow, allowing for robust and fault-tolerant applications.

+ * + * @see CompletableFuture + * @see ApplyFunction + * @see AsyncApplyFunction + * @see AsyncRun + * @see AsyncForEachRun + * @see CatchFunction + * @see AsyncCatchFunction + * @see FinallyFunction + * @see AsyncBiFunction + */ +public final class AsyncUtil { + private static final Boolean BOOLEAN_RESULT = false; + private static final Long LONG_RESULT = -1L; + private static final Object NULL_RESULT = null; + + private AsyncUtil(){} + + /** + * Provides a default value based on the type specified. + * + * @param clazz The {@link Class} object representing the type of the value + * to be returned. + * @param The type of the value to be returned. + * @return An object with a value determined by the type: + *
    + *
  • {@code false} if {@code clazz} is {@link Boolean}, + *
  • -1 if {@code clazz} is {@link Long}, + *
  • {@code null} for any other type. + *
+ */ + public static R asyncReturn(Class clazz) { + if (clazz == null) { + return null; + } + if (clazz.equals(Boolean.class)) { + return (R) BOOLEAN_RESULT; + } else if (clazz.equals(Long.class)) { + return (R) LONG_RESULT; + } + return (R) NULL_RESULT; + } + + /** + * Synchronously returns the result of the current asynchronous operation. + * This method is designed to be used in scenarios where the result of an + * asynchronous operation is needed synchronously, and it is known that + * the operation has completed. + * + *

The method retrieves the current thread's {@link CompletableFuture} and + * attempts to get the result. If the future is not yet complete, this + * method will block until the result is available. If the future completed + * exceptionally, the cause of the exception is thrown as a runtime + * exception wrapped in an {@link ExecutionException}.

+ * + *

This method is typically used after an asynchronous operation has been + * initiated and the caller needs to obtain the result in a synchronous + * manner, for example, when bridging between asynchronous and synchronous + * code paths.

+ * + * @param the type of the result to be returned + * @param clazz the {@link Class} object representing the type of the value + * to be returned, used to cast the result to the correct type + * @return the result of the asynchronous operation as an object of the + * specified class + * @throws Exception if an error occurs during the synchronous retrieval of + * the result, including the original exception if the + * future completed exceptionally + */ + public static R syncReturn(Class clazz) + throws Exception { + CompletableFuture completableFuture = CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + try { + return (R) completableFuture.get(); + } catch (ExecutionException e) { + throw (Exception)e.getCause(); + } + } + + /** + * Completes the current asynchronous operation with the specified value. + * This method sets the result of the current thread's {@link CompletableFuture} + * to the provided value, effectively completing the asynchronous operation. + * + * @param value The value to complete the future with. + * @param The type of the value to be completed. + */ + public static void asyncComplete(R value) { + CUR_COMPLETABLE_FUTURE.set( + CompletableFuture.completedFuture(value)); + } + + /** + * Completes the current asynchronous operation with an exception. + * This method sets the result of the current thread's {@link CompletableFuture} + * to an exceptional completion, using the provided {@link Throwable} as the cause. + * This is typically used to handle errors in asynchronous operations. + * + * @param e The exception to complete the future exceptionally with. + */ + public static void asyncThrowException(Throwable e) { + CompletableFuture result = new CompletableFuture<>(); + result.completeExceptionally(warpCompletionException(e)); + CUR_COMPLETABLE_FUTURE.set(result); + } + + /** + * Applies an asynchronous function to the current {@link CompletableFuture}. + * This method retrieves the current thread's {@link CompletableFuture} and applies + * the provided {@link ApplyFunction} to it. It is used to chain asynchronous + * operations, where the result of one operation is used as the input for the next. + * + * @param function The asynchronous function to apply, which takes a type T and + * produces a type R. + * @param The type of the input to the function. + * @param The type of the result of the function. + * @see CompletableFuture + * @see ApplyFunction + */ + public static void asyncApply(ApplyFunction function) { + CompletableFuture completableFuture = + (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + CompletableFuture result = function.apply(completableFuture); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } + + /** + * Applies an asynchronous function to the current {@link CompletableFuture} + * using the specified executor. This method retrieves the current thread's + * {@link CompletableFuture} and applies the provided{@link ApplyFunction} to + * it with the given executor service. It allows for more control over the + * execution context, such as running the operation in a separate thread or + * thread pool. + * + *

This is particularly useful when you need to perform blocking I/O operations + * or other long-running tasks without blocking the main thread or + * when you want to manage the thread resources more efficiently.

+ * + * @param function The asynchronous function to apply, which takes a type T and + * produces a type R. + * @param executor The executor service used to run the asynchronous function. + * @param The type of the input to the function. + * @param The type of the result of the function. + * @see CompletableFuture + * @see ApplyFunction + */ + public static void asyncApplyUseExecutor( + ApplyFunction function, Executor executor) { + CompletableFuture completableFuture = + (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + CompletableFuture result = function.apply(completableFuture, executor); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } + + /** + * Attempts to execute an asynchronous task defined by the provided + * {@link AsyncRun} and associates it with the current thread's + * {@link CompletableFuture}. This method is useful for trying operations + * that may throw exceptions and handling them asynchronously. + * + *

The provided {@code asyncRun} is a functional interface that + * encapsulates the logic to be executed asynchronously. It is executed in + * the context of the current CompletableFuture, allowing for chaining further + * asynchronous operations based on the result or exception of this try.

+ * + *

If the operation completes successfully, the result is propagated to the + * next operation in the chain. If an exception occurs, it can be caught and + * handled using the {@link #asyncCatch(CatchFunction, Class)} method, + * allowing for error recovery or alternative processing.

+ * + * @param asyncRun The asynchronous task to be executed, defined by + * an {@link AsyncRun} instance. + * @param The type of the result produced by the asynchronous task. + * @see AsyncRun + * @see #asyncCatch(CatchFunction, Class) + */ + public static void asyncTry(AsyncRun asyncRun) { + try { + CompletableFuture result = asyncRun.async(); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } catch (Throwable e) { + asyncThrowException(e); + } + } + + /** + * Handles exceptions to a specified type that may occur during + * an asynchronous operation. This method is used to catch and deal + * with exceptions in a non-blocking manner, allowing the application + * to continue processing even when errors occur. + * + *

The provided {@code function} is a {@link CatchFunction} that + * defines how to handle the caught exception. It takes the result of + * the asynchronous operation (if any) and the caught exception, and + * returns a new result or modified result to continue the asynchronous + * processing.

+ * + *

The {@code eClass} parameter specifies the type of exceptions to + * catch. Only exceptions that are instances of this type (or its + * subclasses) will be caught and handled by the provided function.

+ * + * @param function The {@link CatchFunction} that defines how to + * handle the caught exception. + * @param eClass The class of the exception type to catch. + * @param The type of the result of the asynchronous operation. + * @param The type of the exception to catch. + * @see CatchFunction + */ + public static void asyncCatch( + CatchFunction function, Class eClass) { + CompletableFuture completableFuture = + (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + CompletableFuture result = function.apply(completableFuture, eClass); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } + + /** + * Executes a final action after an asynchronous operation + * completes, regardless of whether the operation was successful + * or resulted in an exception. This method provides a way to + * perform cleanup or finalization tasks in an asynchronous + * workflow. + * + *

The provided {@code function} is a {@link FinallyFunction} + * that encapsulates the logic to be executed after the + * asynchronous operation. It takes the result of the operation + * and returns a new result, which can be used to continue the + * asynchronous processing or to handle the final output of + * the workflow.

+ * + *

This method is particularly useful for releasing resources, + * closing connections, or performing other cleanup actions that + * need to occur after all other operations have completed.

+ * + * @param function The {@link FinallyFunction} that defines + * the final action to be executed. + * @param The type of the result of the asynchronous + * operation. + * @see FinallyFunction + */ + public static void asyncFinally(FinallyFunction function) { + CompletableFuture completableFuture = + (CompletableFuture) CUR_COMPLETABLE_FUTURE.get(); + assert completableFuture != null; + CompletableFuture result = function.apply(completableFuture); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } + + /** + * Executes an asynchronous operation for each element in an Iterator, applying + * a given async function to each element. This method is part of the asynchronous + * utilities provided to facilitate non-blocking operations on collections of elements. + * + *

The provided {@code asyncDo} is an {@link AsyncBiFunction} that encapsulates + * the logic to be executed asynchronously for each element. It is executed in + * the context of the current CompletableFuture, allowing for chaining further + * asynchronous operations based on the result or exception of each iteration.

+ * + *

The method is particularly useful for performing asynchronous iterations + * over collections where the processing of each element is independent.

+ * + * @param forEach the Iterator over which to iterate and apply the async function + * @param asyncDo the asynchronous function to apply to each element of the Iterator, + * implemented as an {@link AsyncBiFunction} + * @param the type of the elements being iterated over + * @param the type of the result produced by the asynchronous task applied to each element + * @see AsyncBiFunction + * @see AsyncForEachRun + */ + public static void asyncForEach( + Iterator forEach, AsyncBiFunction, I, R> asyncDo) { + AsyncForEachRun asyncForEachRun = new AsyncForEachRun<>(); + asyncForEachRun.forEach(forEach).asyncDo(asyncDo).run(); + } + + /** + * Applies an asynchronous operation to each element of a collection + * and aggregates the results. This method is designed to process a + * collection of elements concurrently using asynchronous tasks, and + * then combine the results into a single aggregated result. + * + *

The operation defined by {@code asyncDo} is applied to each + * element of the collection. This operation is expected to return a + * {@link CompletableFuture} representing the asynchronous task. + * Once all tasks have been started, the method (async) waits for all of + * them to complete and then uses the {@code then} function to + * process and aggregate the results.

+ * + *

The {@code then} function takes an array of {@link CompletableFuture} + * instances, each representing the future result of an individual + * asynchronous operation. It should return a new aggregated result + * based on these futures. This allows for various forms of result + * aggregation, such as collecting all results into a list, + * reducing them to a single value, or performing any other custom + * aggregation logic.

+ * + * @param collection the collection of elements to process. + * @param asyncDo the asynchronous operation to apply to each + * element. It must return a {@link CompletableFuture} + * representing the operation. + * @param then a function that takes an array of futures + * representing the results of the asynchronous + * operations and returns an aggregated result. + * @param the type of the elements in the collection. + * @param the type of the intermediate result from the + * asynchronous operations. + * @param

the type of the final aggregated result. + * @see CompletableFuture + */ + public static void asyncCurrent( + Collection collection, AsyncApplyFunction asyncDo, + Function[], P> then) { + CompletableFuture[] completableFutures = + new CompletableFuture[collection.size()]; + int i = 0; + for(I entry : collection) { + CompletableFuture future = null; + try { + future = asyncDo.async(entry); + } catch (IOException e) { + future = new CompletableFuture<>(); + future.completeExceptionally(warpCompletionException(e)); + } + completableFutures[i++] = future; + } + CompletableFuture

result = CompletableFuture.allOf(completableFutures) + .handle((unused, throwable) -> then.apply(completableFutures)); + CUR_COMPLETABLE_FUTURE.set((CompletableFuture) result); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/CatchFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/CatchFunction.java new file mode 100644 index 0000000000000..fbb0af56ce046 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/CatchFunction.java @@ -0,0 +1,120 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.unWarpCompletionException; +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The {@code CatchFunction} interface represents a function that handles exceptions + * occurring within an asynchronous operation. It provides a mechanism to catch and + * process exceptions to a specific type, allowing for error recovery or alternative + * processing paths within an asynchronous workflow. + * + *

This interface is part of the asynchronous utilities provided by the Hadoop + * Distributed File System (HDFS) Federation router. It is used in conjunction with + * other asynchronous interfaces such as {@link AsyncRun} and + * {@link FinallyFunction} to build complex, non-blocking operations.

+ * + *

An implementation of this interface should define how to handle a caught + * exception. It takes two parameters: the result of the asynchronous operation (if + * any) and the caught exception. The function can then return a new result or modify + * the existing result to continue the asynchronous processing.

+ * + *

CatchFunction is used to implement the following semantics:

+ * 
+ * {@code
+ *    try{
+ *      R res = doAsync(input);
+ *    } catch(E e) {
+ *      // Can use CatchFunction
+ *      R result = thenApply(res, e);
+ *    }
+ * }
+ *
+ * + * @param the type of the result of the asynchronous operation + * @param the type of the exception to catch, extending {@link Throwable} + * @see AsyncRun + * @see FinallyFunction + */ +@FunctionalInterface +public interface CatchFunction + extends Async{ + + /** + * Applies this catch function to the given result and exception. + *

+ * This method is called to process an exception that occurred during an asynchronous + * operation. The implementation of this method should define how to handle the + * caught exception. It may involve logging the error, performing a recovery operation, + * or any other custom error handling logic. + *

+ * The method takes two parameters: the result of the asynchronous operation (if any), + * and the caught exception. Depending on the implementation, the method may return a + * new result, modify the existing result, or throw a new exception. + * + * @param r the result of the asynchronous operation, which may be null if the operation + * did not complete successfully + * @param e the caught exception, which the function should handle + * @return the result after applying the catch function, which may be a new result or a + * modified version of the input result + * @throws IOException if an I/O error occurs during the application of the catch function + */ + R apply(R r, E e) throws IOException; + + /** + * Applies the catch function to a {@code CompletableFuture}, handling exceptions of a + * specified type. + *

+ * This default method provides a way to integrate exception handling into a chain of + * asynchronous operations. It takes a {@code CompletableFuture} and a class object + * representing the type of exception to catch. The method uses the handle method of the + * {@code CompletableFuture} to apply the catch function. + *

+ * If the input future completes exceptionally with an instance of the specified exception + * type, the catch function is applied to the exception. If the future completes with a + * different type of exception or normally, the original result or exception is propagated. + * + * @param in the input {@code CompletableFuture} to which the catch function is applied + * @param eClazz the class object representing the exception type to catch + * @return a new {@code CompletableFuture} that completes with the result of applying + * the catch function, or propagates the original exception if it does not match + * the specified type + */ + default CompletableFuture apply( + CompletableFuture in, Class eClazz) { + return in.handle((r, e) -> { + if (e == null) { + return r; + } + Throwable readException = unWarpCompletionException(e); + if (eClazz.isInstance(readException)) { + try { + return CatchFunction.this.apply(r, (E) readException); + } catch (IOException ioe) { + throw warpCompletionException(ioe); + } + } + throw warpCompletionException(e); + }); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/FinallyFunction.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/FinallyFunction.java new file mode 100644 index 0000000000000..0243f0a0a1a7f --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/FinallyFunction.java @@ -0,0 +1,96 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +import static org.apache.hadoop.hdfs.server.federation.router.async.Async.warpCompletionException; + +/** + * The {@code FinallyFunction} interface represents a function that is used to perform + * final actions after an asynchronous operation completes, regardless of whether the + * operation was successful or resulted in an exception. This interface is part of + * the asynchronous utilities provided by the Hadoop Distributed File System (HDFS) + * Federation router. + * + *

A {@code FinallyFunction} is typically used for cleanup or finalization tasks, + * such as releasing resources, closing connections, or performing other actions that + * need to occur after all other operations have completed.

+ * + *

An implementation of this interface should define what the final action is. It + * takes the result of the asynchronous operation as an argument and returns a new + * result, which can be the same as the input result or a modified version of it.

+ * + *

FinallyFunction is used to implement the following semantics:

+ * 
+ * {@code
+ *    try{
+ *      R res = doAsync(input);
+ *    } catch(...) {
+ *      ...
+ *    } finally {
+ *      // Can use FinallyFunction
+ *      R result = thenApply(res);
+ *    }
+ * }
+ *
+ * + * @param the type of the result of the asynchronous operation + */ +@FunctionalInterface +public interface FinallyFunction { + + /** + * Applies this final action function to the result of an asynchronous operation. + * + * @param r the result of the asynchronous operation, which may be null if the + * operation did not complete successfully + * @return the result after applying the final action, which may be a new result or a + * modified version of the input result + * @throws IOException if an I/O error occurs during the application of the final action + */ + R apply(R r) throws IOException; + + /** + * Applies this final action function to a {@code CompletableFuture}, which is expected + * to be the result of an asynchronous operation. + *

+ * This method is a convenience that simplifies the use of {@code FinallyFunction} + * with asynchronous operations. It handles the completion of the future and applies + * the {@code FinallyFunction} to the result. + * + * @param in the {@code CompletableFuture} representing the asynchronous operation + * @return a new {@code CompletableFuture} that completes with the result of applying + * the final action function + */ + default CompletableFuture apply(CompletableFuture in) { + return in.handle((r, e) -> { + try { + R ret = apply(r); + if (e != null) { + throw warpCompletionException(e); + } else { + return ret; + } + } catch (IOException ioe) { + throw warpCompletionException(ioe); + } + }); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/package-info.java new file mode 100644 index 0000000000000..48fd0ad89abfe --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/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. + */ + +/** + * This package contains classes that facilitate asynchronous operations within the Hadoop + * Distributed File System (HDFS) Federation router. These classes are designed to work with + * the Hadoop ecosystem, providing utilities and interfaces to perform non-blocking tasks that + * can improve the performance and responsiveness of HDFS operations. + * + *

These classes work together to enable complex asynchronous workflows, making it easier to + * write code that can handle long-running tasks without blocking, thus improving the overall + * efficiency and scalability of HDFS operations.

+ */ +@InterfaceAudience.Private +@InterfaceStability.Evolving + +package org.apache.hadoop.hdfs.server.federation.router.async; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncClass.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncClass.java new file mode 100644 index 0000000000000..e5bf7ce08e88c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/AsyncClass.java @@ -0,0 +1,249 @@ +/** + * 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.hdfs.server.federation.router.async; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.Function; + +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncApply; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncCatch; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncComplete; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncCurrent; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncFinally; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncForEach; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncReturn; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncThrowException; +import static org.apache.hadoop.hdfs.server.federation.router.async.AsyncUtil.asyncTry; + +/** + * AsyncClass demonstrates the conversion of synchronous methods + * from SyncClass into asynchronous operations using AsyncUtil. + * This class overrides methods with asynchronous logic, enhancing + * the performance by allowing non-blocking task execution. + * + *

+ * By utilizing AsyncUtil's utility methods, such as asyncApply, + * asyncForEach, and others, each method in AsyncClass can perform + * time-consuming tasks on a separate thread, thus not blocking + * the main execution thread. + *

+ * + *

+ * For example, the applyMethod in AsyncClass is an async version of + * the same method in SyncClass. It uses asyncApply to schedule + * the timeConsumingMethod to run asynchronously and returns a + * CompletableFuture that will be completed with the result of + * the operation. + *

+ * + *

+ * This class serves as an example of how to transform synchronous + * operations into asynchronous ones using the AsyncUtil tools, + * which can be applied to other parts of the HDFS Federation + * router or similar systems to improve concurrency and + * performance. + *

+ * + * @see SyncClass + * @see AsyncUtil + * @see CompletableFuture + */ +public class AsyncClass extends SyncClass{ + private static final Logger LOG = + LoggerFactory.getLogger(AsyncClass.class); + private ExecutorService executorService; + private final static String ASYNC_WORKER = "Async Worker"; + + public AsyncClass(long timeConsuming) { + super(timeConsuming); + executorService = Executors.newFixedThreadPool(1, r -> { + Thread asyncWork = new Thread(r); + asyncWork.setDaemon(true); + asyncWork.setName(ASYNC_WORKER); + return asyncWork; + }); + } + + @Override + public String applyMethod(int input) { + timeConsumingMethod(input); + asyncApply(res -> { + return "applyMethod" + res; + }); + return asyncReturn(String.class); + } + + @Override + public String applyMethod(int input, boolean canException) { + timeConsumingMethod(input); + asyncApply(res -> { + if (canException) { + if (res.equals("[2]")) { + throw new IOException("input 2 exception"); + } else if (res.equals("[3]")) { + throw new RuntimeException("input 3 exception"); + } + } + return res; + }); + return asyncReturn(String.class); + } + + @Override + public String exceptionMethod(int input) { + if (input == 2) { + asyncThrowException(new IOException("input 2 exception")); + return null; + } else if (input == 3) { + asyncThrowException(new RuntimeException("input 3 exception")); + return null; + } + return applyMethod(input); + } + + @Override + public String forEachMethod(List list) { + StringBuilder result = new StringBuilder(); + asyncForEach(list.iterator(), + (forEach, input) -> { + timeConsumingMethod(input); + asyncApply(res -> { + result.append("forEach" + res + ","); + return result.toString(); + }); + }); + return asyncReturn(String.class); + } + + @Override + public String forEachBreakMethod(List list) { + StringBuilder result = new StringBuilder(); + asyncForEach(list.iterator(), + (forEach, input) -> { + timeConsumingMethod(input); + asyncApply(res -> { + if (res.equals("[2]")) { + forEach.breakNow(); + } else { + result.append("forEach" + res + ","); + } + return result.toString(); + }); + }); + return asyncReturn(String.class); + } + + @Override + public String forEachBreakByExceptionMethod(List list) { + StringBuilder result = new StringBuilder(); + asyncForEach(list.iterator(), + (forEach, input) -> { + asyncTry(() -> { + applyMethod(input, true); + asyncApply(res -> { + result.append("forEach" + res + ","); + return result.toString(); + }); + }); + asyncCatch((res, e) -> { + if (e instanceof IOException) { + result.append(e + ","); + } else if (e instanceof RuntimeException) { + forEach.breakNow(); + } + return result.toString(); + }, Exception.class); + }); + return asyncReturn(String.class); + } + + @Override + public String applyThenApplyMethod(int input) { + timeConsumingMethod(input); + asyncApply((AsyncApplyFunction) res -> { + if (res.equals("[1]")) { + timeConsumingMethod(2); + } else { + asyncComplete(res); + } + }); + return asyncReturn(String.class); + } + + @Override + public String applyCatchThenApplyMethod(int input) { + asyncTry(() -> applyMethod(input, true)); + asyncCatch((AsyncCatchFunction) (res, ioe) -> { + applyMethod(1); + }, IOException.class); + return asyncReturn(String.class); + } + + @Override + public String applyCatchFinallyMethod( + int input, List resource) { + asyncTry(() -> applyMethod(input, true)); + asyncCatch((res, e) -> { + throw new IOException("Catch " + e.getMessage()); + }, IOException.class); + asyncFinally((FinallyFunction) res -> { + resource.clear(); + return res; + }); + return asyncReturn(String.class); + } + + @Override + public String currentMethod(List list) { + asyncCurrent(list, + input -> applyMethod(input, true), + (Function[], String>) futures -> { + StringBuilder result = new StringBuilder(); + for (Future future : futures) { + try { + String res = future.get(); + result.append(res + ","); + } catch (Exception e) { + result.append(e.getMessage() + ","); + } + } + return result.toString(); + }); + return asyncReturn(String.class); + } + + @Override + public String timeConsumingMethod(int input) { + CompletableFuture result = CompletableFuture + .supplyAsync(() -> { + LOG.info("[{} thread] invoke consumingMethod for parameter: {}", + Thread.currentThread().getName(), input); + return AsyncClass.super.timeConsumingMethod(input); + }, executorService); + Async.CUR_COMPLETABLE_FUTURE.set(result); + return null; + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/BaseClass.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/BaseClass.java new file mode 100644 index 0000000000000..8d5b5b1dc82ce --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/BaseClass.java @@ -0,0 +1,66 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.List; + +/** + * It defines a set of methods that can be executed either synchronously + * or asynchronously, depending on the implementation. + * + *

+ * This interface is designed to abstract the common operations that need + * to be performed in a time-consuming manner, such as processing a list + * of items or applying a method that involves I/O operations. By defining + * these methods in an interface, it allows for both synchronous and + * asynchronous implementations, providing flexibility and the ability to + * improve performance without changing the external API. + *

+ * + *

+ * Implementations of this interface are expected to provide concrete + * implementations of the defined methods, either by performing the + * operations synchronously in a blocking manner or by performing them + * asynchronously in a non-blocking manner. + *

+ * + * @see SyncClass + * @see AsyncClass + */ +public interface BaseClass { + String applyMethod(int input); + + String applyMethod(int input, boolean canException) throws IOException; + + String exceptionMethod(int input) throws IOException; + + String forEachMethod(List list); + + String forEachBreakMethod(List list); + + String forEachBreakByExceptionMethod(List list); + + String applyThenApplyMethod(int input); + + String applyCatchThenApplyMethod(int input); + + String applyCatchFinallyMethod(int input, List resource) throws IOException; + + String currentMethod(List list); +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/SyncClass.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/SyncClass.java new file mode 100644 index 0000000000000..e55edb098e116 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/SyncClass.java @@ -0,0 +1,194 @@ +/** + * 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.hdfs.server.federation.router.async; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +/** + * SyncClass implements BaseClass, providing a synchronous + * version of the methods. All operations are performed in a + * blocking manner, waiting for completion before proceeding. + * + * This class is the foundation for the AsyncClass, which + * provides asynchronous implementations. + * + * @see BaseClass + * @see AsyncClass + */ +public class SyncClass implements BaseClass{ + private long timeConsuming; + + public SyncClass(long timeConsuming) { + this.timeConsuming = timeConsuming; + } + + @Override + public String applyMethod(int input) { + String res = timeConsumingMethod(input); + return "applyMethod" + res; + } + + @Override + public String applyMethod(int input, boolean canException) throws IOException { + String res = timeConsumingMethod(input); + if (canException) { + if (res.equals("[2]")) { + throw new IOException("input 2 exception"); + } else if (res.equals("[3]")) { + throw new RuntimeException("input 3 exception"); + } + } + return res; + } + + @Override + public String exceptionMethod(int input) throws IOException { + if (input == 2) { + throw new IOException("input 2 exception"); + } else if (input == 3) { + throw new RuntimeException("input 3 exception"); + } + return applyMethod(input); + } + + @Override + public String forEachMethod(List list) { + StringBuilder result = new StringBuilder(); + for (int input : list) { + String res = timeConsumingMethod(input); + result.append("forEach" + res + ","); + } + return result.toString(); + } + + @Override + public String forEachBreakMethod(List list) { + StringBuilder result = new StringBuilder(); + for (int input : list) { + String res = timeConsumingMethod(input); + if (res.equals("[2]")) { + break; + } + result.append("forEach" + res + ","); + } + return result.toString(); + } + + @Override + public String forEachBreakByExceptionMethod(List list) { + StringBuilder result = new StringBuilder(); + for (int input : list) { + try { + String res = applyMethod(input, true); + result.append("forEach" + res + ","); + } catch (IOException e) { + result.append(e + ","); + } catch (RuntimeException e) { + break; + } + } + return result.toString(); + } + + @Override + public String applyThenApplyMethod(int input) { + String res = timeConsumingMethod(input); + if (res.equals("[1]")) { + res = timeConsumingMethod(2); + } + return res; + } + + @Override + public String applyCatchThenApplyMethod(int input) { + String res = null; + try { + res = applyMethod(input, true); + } catch (IOException e) { + res = applyMethod(1); + } + return res; + } + + @Override + public String applyCatchFinallyMethod( + int input, List resource) throws IOException { + String res = null; + try { + res = applyMethod(input, true); + } catch (IOException e) { + throw new IOException("Catch " + e.getMessage()); + } finally { + resource.clear(); + } + return res; + } + + @Override + public String currentMethod(List list) { + ExecutorService executor = getExecutorService(); + List> futures = new ArrayList<>(); + for (int input : list) { + Future future = executor.submit( + () -> applyMethod(input, true)); + futures.add(future); + } + + StringBuilder result = new StringBuilder(); + for (Future future : futures) { + try { + String res = future.get(); + result.append(res + ","); + } catch (Exception e) { + result.append(e.getMessage() + ","); + } + } + return result.toString(); + } + + + /** + * Simulates a synchronous method that performs + * a time-consuming task and returns a result. + * + * @param input The input parameter for the method. + * @return A string that represents the result of the method. + */ + public String timeConsumingMethod(int input) { + try { + Thread.sleep(timeConsuming); + return "[" + input + "]"; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return "Error:" + e.getMessage(); + } + } + + private ExecutorService getExecutorService() { + return Executors.newFixedThreadPool(2, r -> { + Thread t = new Thread(r); + t.setDaemon(true); + return t; + }); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestAsyncUtil.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestAsyncUtil.java new file mode 100644 index 0000000000000..c540af612b974 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestAsyncUtil.java @@ -0,0 +1,277 @@ +/** + * 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.hdfs.server.federation.router.async; + +import org.apache.hadoop.test.LambdaTestUtils; +import org.apache.hadoop.util.Time; +import org.junit.After; +import org.junit.Before; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Callable; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * The TestAsyncUtil class provides a suite of test cases for the + * asynchronous utility class AsyncUtil. It utilizes the JUnit testing + * framework to verify that asynchronous operations are performed as + * expected. + * + *

+ * This class contains multiple test methods designed to test various + * asynchronous operation scenarios, including: + *

    + *
  • testApply - Tests the asynchronous application of a method.
    • + *
  • testApplyException - Tests exception handling in + * asynchronous methods.
    • + *
  • testApplyThenApplyMethod - Tests the chaining of + * asynchronous method calls.
    • + *
  • testCatchThenApplyMethod - Tests the invocation of + * asynchronous methods after exception catching.
    • + *
  • testForEach - Tests asynchronous iteration operations.
    • + *
  • testForEachBreak - Tests asynchronous iteration with break + * conditions.
    • + *
  • testForEachBreakByException - Tests the interruption of + * asynchronous iteration due to exceptions.
    • + *
+ *

+ * + * The tests cover both synchronous (Sync) and asynchronous (Async) + * configurations to ensure consistent behavior under different + * execution modes. + * + * @see AsyncUtil + * @see BaseClass + * @see SyncClass + * @see AsyncClass + */ +public class TestAsyncUtil { + private static final Logger LOG = + LoggerFactory.getLogger(TestAsyncUtil.class); + private static final long TIME_CONSUMING = 100; + private BaseClass baseClass; + private boolean enableAsync; + + public enum ExecutionMode { + SYNC, + ASYNC + } + + @Before + public void setUp(ExecutionMode mode) { + if (mode.equals(ExecutionMode.ASYNC)) { + baseClass = new AsyncClass(TIME_CONSUMING); + enableAsync = true; + } else { + baseClass = new SyncClass(TIME_CONSUMING); + } + } + + @After + public void after() { + baseClass = null; + enableAsync = false; + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testApply(ExecutionMode mode) + throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.applyMethod(1); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("applyMethod[1]", result, TIME_CONSUMING, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testApplyException(ExecutionMode mode) throws Exception { + setUp(mode); + checkException( + () -> baseClass.applyMethod(2, true), + IOException.class, "input 2 exception"); + + checkException( + () -> baseClass.applyMethod(3, true), + RuntimeException.class, "input 3 exception"); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testExceptionMethod(ExecutionMode mode) throws Exception { + setUp(mode); + checkException( + () -> baseClass.exceptionMethod(2), + IOException.class, "input 2 exception"); + + checkException( + () -> baseClass.exceptionMethod(3), + RuntimeException.class, "input 3 exception"); + + long start = Time.monotonicNow(); + String result = baseClass.exceptionMethod(1); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("applyMethod[1]", result, TIME_CONSUMING, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testApplyThenApplyMethod(ExecutionMode mode) throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.applyThenApplyMethod(1); + long cost = Time.monotonicNow() - start; + checkResult("[2]", result, TIME_CONSUMING, cost); + LOG.info("[{}] main thread cost: {} ms", mode, cost); + + start = Time.monotonicNow(); + result = baseClass.applyThenApplyMethod(3); + cost = Time.monotonicNow() - start; + checkResult("[3]", result, TIME_CONSUMING, cost); + LOG.info("[{}] main thread cost: {} ms", mode, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testCatchThenApplyMethod(ExecutionMode mode) throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.applyCatchThenApplyMethod(2); + long cost = Time.monotonicNow() - start; + checkResult("applyMethod[1]", result, TIME_CONSUMING, cost); + LOG.info("[{}] main thread cost: {} ms", mode, cost); + + start = Time.monotonicNow(); + result = baseClass.applyCatchThenApplyMethod(0); + cost = Time.monotonicNow() - start; + checkResult("[0]", result, TIME_CONSUMING, cost); + LOG.info("[{}] main thread cost: {} ms", mode, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testCatchFinallyMethod(ExecutionMode mode) throws Exception { + setUp(mode); + List resource = new ArrayList<>(); + resource.add("resource1"); + checkException( + () -> baseClass.applyCatchFinallyMethod(2, resource), + IOException.class, "input 2 exception"); + assertTrue(resource.size() == 0); + + long start = Time.monotonicNow(); + String result = baseClass.applyCatchFinallyMethod(0, resource); + long cost = Time.monotonicNow() - start; + checkResult("[0]", result, TIME_CONSUMING, cost); + assertTrue(resource.size() == 0); + LOG.info("[{}] main thread cost: {} ms", mode, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testForEach(ExecutionMode mode) throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.forEachMethod(Arrays.asList(1, 2, 3)); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("forEach[1],forEach[2],forEach[3],", result, + TIME_CONSUMING, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testForEachBreak(ExecutionMode mode) throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.forEachBreakMethod(Arrays.asList(1, 2, 3)); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("forEach[1],", result, TIME_CONSUMING, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testForEachBreakByException(ExecutionMode mode) + throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.forEachBreakByExceptionMethod(Arrays.asList(1, 2, 3)); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("forEach[1],java.io.IOException: input 2 exception,", + result, TIME_CONSUMING, cost); + } + + @EnumSource(ExecutionMode.class) + @ParameterizedTest + public void testCurrentMethod(ExecutionMode mode) + throws Exception { + setUp(mode); + long start = Time.monotonicNow(); + String result = baseClass.currentMethod(Arrays.asList(1, 2, 3)); + long cost = Time.monotonicNow() - start; + LOG.info("[{}] main thread cost: {} ms", mode, cost); + checkResult("[1],java.io.IOException: input 2 exception," + + "java.lang.RuntimeException: input 3 exception,", + result, TIME_CONSUMING, cost); + } + + private void checkResult( + String result, String actualResult, long cost, long actualCost) + throws Exception { + if (enableAsync) { + Assertions.assertNull(actualResult); + actualResult = AsyncUtil.syncReturn(String.class); + assertNotNull(actualResult); + assertTrue(actualCost < cost); + } else { + assertFalse(actualCost < cost); + } + assertEquals(result, actualResult); + } + + private < E extends Throwable> void checkException( + Callable eval, Class clazz, String contained) throws Exception { + if (enableAsync) { + LambdaTestUtils.intercept(clazz, contained, + () -> { + eval.call(); + return AsyncUtil.syncReturn(String.class); + }); + } else { + LambdaTestUtils.intercept(clazz, contained, () -> { + String res = eval.call(); + return res; + }); + } + } +}