You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by st...@apache.org on 2023/05/22 09:38:46 UTC
[hadoop] 06/12: Add unit tests for BlockingEnumeration
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch feature-HADOOP-18073-s3a-sdk-upgrade
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 38dff867b9d942830a222a493a45b686d2249705
Author: Alessandro Passaro <al...@amazon.co.uk>
AuthorDate: Mon Dec 5 11:39:28 2022 +0000
Add unit tests for BlockingEnumeration
---
.../hadoop/fs/s3a/select/BlockingEnumeration.java | 11 +-
.../hadoop/fs/s3a/select/StreamPublisher.java | 89 +++++++++
.../fs/s3a/select/TestBlockingEnumeration.java | 200 +++++++++++++++++++++
3 files changed, 297 insertions(+), 3 deletions(-)
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java
index c09acc8ba22..bc47db47c76 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java
@@ -96,13 +96,18 @@ public final class BlockingEnumeration<T> implements Enumeration<T> {
} catch (InterruptedException e) {
current = new Signal<>(e);
subscription.thenAccept(Subscription::cancel);
+ Thread.currentThread().interrupt();
}
}
if (current.error != null) {
- if (current.error instanceof SdkException) {
- throw (SdkException)current.error;
+ Throwable error = current.error;
+ current = END_SIGNAL;
+ if (error instanceof Error) {
+ throw (Error)error;
+ } else if (error instanceof SdkException) {
+ throw (SdkException)error;
} else {
- throw SdkException.create("Unexpected error", current.error);
+ throw SdkException.create("Unexpected error", error);
}
}
return current != END_SIGNAL;
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.java
new file mode 100644
index 00000000000..c770b889733
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.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.fs.s3a.select;
+
+import java.util.Iterator;
+import java.util.concurrent.Executor;
+import java.util.stream.Stream;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.core.async.SdkPublisher;
+
+/**
+ * Publisher used to test the handling of asynchronous responses.
+ * @param <T> The type of published elements.
+ */
+final class StreamPublisher<T> implements SdkPublisher<T> {
+ private final Executor executor;
+ private final Iterator<T> iterator;
+ private Boolean done = false;
+
+ public StreamPublisher(Stream<T> data, Executor executor) {
+ this.iterator = data.iterator();
+ this.executor = executor;
+ }
+
+ public StreamPublisher(Stream<T> data) {
+ this(data, Runnable::run);
+ }
+
+ @Override
+ public void subscribe(Subscriber<? super T> subscriber) {
+ subscriber.onSubscribe(new Subscription() {
+ @Override
+ public void request(long n) {
+ if (done) {
+ return;
+ }
+
+ if (n < 1) {
+ done = true;
+ executor.execute(() -> subscriber.onError(new IllegalArgumentException()));
+ return;
+ }
+
+ for (long i = 0; i < n; i++) {
+ final T value;
+ try {
+ synchronized (iterator) {
+ value = iterator.hasNext() ? iterator.next() : null;
+ }
+ } catch (Throwable e) {
+ executor.execute(() -> subscriber.onError(e));
+ break;
+ }
+
+ if (value == null) {
+ done = true;
+ executor.execute(subscriber::onComplete);
+ break;
+ } else {
+ executor.execute(() -> subscriber.onNext(value));
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ done = true;
+ }
+ });
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java
new file mode 100644
index 00000000000..43bdcb062f0
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java
@@ -0,0 +1,200 @@
+/*
+ * 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.fs.s3a.select;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import software.amazon.awssdk.core.exception.SdkException;
+
+/**
+ * Unit tests for {@link BlockingEnumeration}.
+ */
+public final class TestBlockingEnumeration extends Assert {
+
+ @Test
+ public void containsElement() {
+ SdkPublisher<String> publisher = new StreamPublisher<>(Stream.of("foo"));
+
+ BlockingEnumeration<String> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+
+ assertTrue(enumeration.hasMoreElements());
+ assertEquals("foo", enumeration.nextElement());
+ assertFalse(enumeration.hasMoreElements());
+ }
+
+ @Test
+ public void containsInjectedElement() {
+ SdkPublisher<String> publisher = new StreamPublisher<>(Stream.of("foo"));
+
+ BlockingEnumeration<String> enumeration =
+ new BlockingEnumeration<>(publisher, 1, "bar");
+
+ assertTrue(enumeration.hasMoreElements());
+ assertEquals("bar", enumeration.nextElement());
+ assertTrue(enumeration.hasMoreElements());
+ assertEquals("foo", enumeration.nextElement());
+ assertFalse(enumeration.hasMoreElements());
+ }
+
+ @Test
+ public void throwsExceptionOnFirstElement() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .map(i -> {
+ throw SdkException.create("error!", null);
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+ assertThrows(SdkException.class, enumeration::hasMoreElements);
+ }
+
+ @Test
+ public void throwsExceptionAfterInjectedElement() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .peek(i -> {
+ throw SdkException.create("error!", null);
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1, 99);
+ assertTrue(enumeration.hasMoreElements());
+ assertEquals(99, enumeration.nextElement().intValue());
+ assertThrows(SdkException.class, enumeration::hasMoreElements);
+ }
+
+ @Test
+ public void throwsNonSdkException() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .peek(i -> {
+ throw new RuntimeException("error!", null);
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+ SdkException exception = Assert.assertThrows(SdkException.class, enumeration::hasMoreElements);
+ assertEquals(RuntimeException.class, exception.getCause().getClass());
+ }
+
+ @Test
+ public void throwsError() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .peek(i -> {
+ throw new Error("error!", null);
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+ assertThrows(Error.class, enumeration::hasMoreElements);
+ }
+
+ @Test
+ public void throwsExceptionOnSecondElement() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .peek(i -> {
+ if (i == 1) {
+ throw SdkException.create("error!", null);
+ }
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+ assertTrue(enumeration.hasMoreElements());
+ assertEquals(0, enumeration.nextElement().intValue());
+ assertThrows(SdkException.class, enumeration::hasMoreElements);
+ }
+
+ @Test
+ public void noMoreElementsAfterThrow() {
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ Stream.of(0, 1)
+ .map(i -> {
+ throw SdkException.create("error!", null);
+ }),
+ Executors.newSingleThreadExecutor());
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, 1);
+ assertThrows(SdkException.class, enumeration::hasMoreElements);
+ assertFalse(enumeration.hasMoreElements());
+ }
+
+ @Test
+ public void buffersOnSameThread() {
+ verifyBuffering(10, 3, Runnable::run);
+ }
+
+ @Test
+ public void publisherOnDifferentThread() {
+ verifyBuffering(5, 1, Executors.newSingleThreadExecutor());
+ }
+
+ @Test
+ public void publisherOnDifferentThreadWithBuffer() {
+ verifyBuffering(30, 10, Executors.newSingleThreadExecutor());
+ }
+
+ private static void verifyBuffering(int length, int bufferSize, Executor executor) {
+ AtomicInteger emitted = new AtomicInteger();
+ SdkPublisher<Integer> publisher = new StreamPublisher<>(
+ IntStream.range(0, length).boxed().peek(i -> emitted.incrementAndGet()),
+ executor);
+
+ BlockingEnumeration<Integer> enumeration =
+ new BlockingEnumeration<>(publisher, bufferSize);
+
+ int pulled = 0;
+ while (true) {
+ try {
+ int expected = Math.min(length, pulled + bufferSize);
+ if (expected != emitted.get()) {
+ Thread.sleep(10);
+ }
+ assertEquals(expected, emitted.get());
+ } catch (InterruptedException e) {
+ fail("Interrupted: " + e);
+ }
+
+ if (!enumeration.hasMoreElements()) {
+ break;
+ }
+
+ int i = enumeration.nextElement();
+ assertEquals(pulled, i);
+ pulled++;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org