diff --git a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterProvider.java b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterProvider.java index a7373e5322cf..c69ecc37ef66 100644 --- a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterProvider.java +++ b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterProvider.java @@ -125,7 +125,7 @@ public class MiniOzoneClusterProvider { private final Thread reapThread; private final Set createdClusters = new HashSet<>(); - private final BlockingQueue clusters + private final BlockingQueue clusterResults = new ArrayBlockingQueue<>(PRE_CREATE_LIMIT); private final BlockingQueue expiredClusters = new ArrayBlockingQueue<>(EXPIRED_LIMIT); @@ -152,10 +152,14 @@ public synchronized MiniOzoneCluster provide() + "been reached for this provider. Please increase the value set " + "in the constructor"); } - MiniOzoneCluster cluster = clusters.poll(100, SECONDS); - if (cluster == null) { + ClusterCreationResult result = clusterResults.poll(100, SECONDS); + if (result == null) { throw new IOException("Failed to obtain available cluster in time"); } + if (result.getFailure() != null) { + throw result.getFailure(); + } + MiniOzoneCluster cluster = result.getCluster(); createdClusters.add(cluster); consumedClusterCount++; return cluster; @@ -215,14 +219,22 @@ private Thread createClusters() { cluster = builder.build(); cluster.waitForClusterToBeReady(); createdCount++; - clusters.put(cluster); + clusterResults.put(ClusterCreationResult.success(cluster)); } catch (InterruptedException e) { if (cluster != null) { cluster.shutdown(); } break; } catch (IOException | TimeoutException e) { - throw new RuntimeException("Unable to build cluster", e); + LOG.warn("Unable to build cluster", e); + if (cluster != null) { + cluster.shutdown(); + } + try { + clusterResults.put(ClusterCreationResult.failure(e)); + } catch (InterruptedException interrupted) { + break; + } } } }); @@ -232,9 +244,10 @@ private Thread createClusters() { } private void destroyRemainingClusters() { - while (!clusters.isEmpty()) { + while (!clusterResults.isEmpty()) { try { - MiniOzoneCluster cluster = clusters.poll(); + ClusterCreationResult result = clusterResults.poll(); + MiniOzoneCluster cluster = result == null ? null : result.getCluster(); if (cluster != null) { destroy(cluster); } @@ -258,4 +271,32 @@ private void destroyRemainingClusters() { createdClusters.clear(); } + private static final class ClusterCreationResult { + private final MiniOzoneCluster cluster; + private final IOException failure; + + private ClusterCreationResult(MiniOzoneCluster cluster, + IOException failure) { + this.cluster = cluster; + this.failure = failure; + } + + private static ClusterCreationResult success(MiniOzoneCluster cluster) { + return new ClusterCreationResult(cluster, null); + } + + private static ClusterCreationResult failure(Exception failure) { + return new ClusterCreationResult(null, + new IOException("Unable to build cluster", failure)); + } + + private MiniOzoneCluster getCluster() { + return cluster; + } + + private IOException getFailure() { + return failure; + } + } + } diff --git a/hadoop-ozone/mini-cluster/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneClusterProvider.java b/hadoop-ozone/mini-cluster/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneClusterProvider.java new file mode 100644 index 000000000000..5abaa88ad167 --- /dev/null +++ b/hadoop-ozone/mini-cluster/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneClusterProvider.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Queue; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.junit.jupiter.api.Test; + +class TestMiniOzoneClusterProvider { + + @Test + void retriesAfterBuildFailure() throws Exception { + IOException failure = new IOException("failed"); + MiniOzoneCluster first = mock(MiniOzoneCluster.class); + MiniOzoneCluster second = mock(MiniOzoneCluster.class); + TestBuilder builder = new TestBuilder(failure, first, second); + MiniOzoneClusterProvider provider = new MiniOzoneClusterProvider(builder, + 2); + + try { + IOException exception = assertThrows(IOException.class, + provider::provide); + assertSame(failure, exception.getCause()); + assertSame(first, provider.provide()); + assertSame(second, provider.provide()); + assertEquals(3, builder.getBuildCount()); + } finally { + provider.shutdown(); + } + } + + @Test + void retriesAfterReadinessTimeoutAndShutsDownCluster() throws Exception { + TimeoutException failure = new TimeoutException("failed"); + MiniOzoneCluster failed = mock(MiniOzoneCluster.class); + doThrow(failure).when(failed).waitForClusterToBeReady(); + MiniOzoneCluster first = mock(MiniOzoneCluster.class); + MiniOzoneCluster second = mock(MiniOzoneCluster.class); + TestBuilder builder = new TestBuilder(failed, first, second); + MiniOzoneClusterProvider provider = new MiniOzoneClusterProvider(builder, + 2); + + try { + IOException exception = assertThrows(IOException.class, + provider::provide); + assertSame(failure, exception.getCause()); + assertSame(first, provider.provide()); + assertSame(second, provider.provide()); + assertEquals(3, builder.getBuildCount()); + verify(failed).shutdown(); + } finally { + provider.shutdown(); + } + } + + private static final class TestBuilder extends MiniOzoneCluster.Builder { + private final Queue results; + private final AtomicInteger buildCount = new AtomicInteger(); + + private TestBuilder(Object... results) { + super(new OzoneConfiguration()); + this.results = new ArrayDeque<>(Arrays.asList(results)); + } + + @Override + public MiniOzoneCluster build() throws IOException { + buildCount.incrementAndGet(); + Object result = results.remove(); + if (result instanceof IOException) { + throw (IOException) result; + } + return (MiniOzoneCluster) result; + } + + private int getBuildCount() { + return buildCount.get(); + } + } +}