Remove the unnamed Pipe constructor and update tests to set a pipe name

Signed-off-by: Matthew Whitehead <[email protected]>

consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/BFTPivotSelectorFromPeers.java

@@ -76,6 +76,12 @@ public BFTPivotSelectorFromPeers(
     LOG.info("Creating pivot block selector for BFT node");
   }
 
+  /**
+   * Determine if our node is a BFT validator node
+   *
+   * @param validatorProvider the validator provider
+   * @return true if we are a validator
+   */
   protected boolean weAreAValidator(final ValidatorProvider validatorProvider) {
     return validatorProvider.nodeIsValidator(nodeKey);
   }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloader.java

@@ -50,7 +50,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@SuppressWarnings("unused")
 public class SnapWorldStateDownloader implements WorldStateDownloader {
 
   private static final Logger LOG = LoggerFactory.getLogger(SnapWorldStateDownloader.class);

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/LoadLocalDataStepTest.java

@@ -51,7 +51,7 @@ public class LoadLocalDataStepTest {
   private final Task<NodeDataRequest> task = new StubTask(request);
 
   private final Pipe<Task<NodeDataRequest>> completedTasks =
-      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "test_pipe");
   private final LoadLocalDataStep loadLocalDataStep =
       new LoadLocalDataStep(
           new WorldStateStorageCoordinator(worldStateKeyValueStorage), new NoOpMetricsSystem());

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/LoadLocalDataStepTest.java

@@ -55,7 +55,7 @@ public class LoadLocalDataStepTest {
   private final Task<SnapDataRequest> task = new StubTask(request);
 
   private final Pipe<Task<SnapDataRequest>> completedTasks =
-      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "test_pipe");
 
   private final SnapSyncProcessState snapSyncState = mock(SnapSyncProcessState.class);
   private final SnapWorldDownloadState downloadState = mock(SnapWorldDownloadState.class);

services/pipeline/src/main/java/org/hyperledger/besu/services/pipeline/Pipe.java

@@ -46,25 +46,6 @@ public class Pipe<T> implements ReadPipe<T>, WritePipe<T> {
   private final AtomicBoolean aborted = new AtomicBoolean();
   private String pipeName = "";
 
-  /**
-   * Instantiates a new Pipe.
-   *
-   * @param capacity the capacity
-   * @param inputCounter the input counter
-   * @param outputCounter the output counter
-   * @param abortedItemCounter the aborted item counter
-   */
-  public Pipe(
-      final int capacity,
-      final Counter inputCounter,
-      final Counter outputCounter,
-      final Counter abortedItemCounter) {
-    queue = new ArrayBlockingQueue<>(capacity);
-    this.inputCounter = inputCounter;
-    this.outputCounter = outputCounter;
-    this.abortedItemCounter = abortedItemCounter;
-  }
-
   /**
    * Instantiates a new Pipe.
    *

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/BatchingReadPipeTest.java

@@ -31,7 +31,8 @@
 
 public class BatchingReadPipeTest {
 
-  private final Pipe<String> source = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+  private final Pipe<String> source =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "source_pipe");
   private final Counter batchCounter = mock(Counter.class);
   private final BatchingReadPipe<String> batchingPipe =
       new BatchingReadPipe<>(source, 3, batchCounter);

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/CompleterStageTest.java

@@ -24,7 +24,8 @@
 
 public class CompleterStageTest {
 
-  private final Pipe<String> pipe = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+  private final Pipe<String> pipe =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "test_pipe");
   private final List<String> output = new ArrayList<>();
   private final CompleterStage<String> stage = new CompleterStage<>("name", pipe, output::add);
 

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/FlatMapProcessorTest.java

@@ -28,8 +28,10 @@
 
 public class FlatMapProcessorTest {
 
-  private final Pipe<String> input = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
-  private final Pipe<String> output = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+  private final Pipe<String> input =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "input_pipe");
+  private final Pipe<String> output =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "output_pipe");
 
   @SuppressWarnings("unchecked")
   private final Function<String, Stream<String>> mapper = mock(Function.class);

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/IteratorSourceStageTest.java

@@ -22,7 +22,8 @@
 
 public class IteratorSourceStageTest {
 
-  private final Pipe<String> output = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+  private final Pipe<String> output =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "output_pipe");
 
   private final IteratorSourceStage<String> stage =
       new IteratorSourceStage<>("name", Iterators.forArray("a", "b", "c", "d"), output);

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/MapProcessorTest.java

@@ -27,8 +27,10 @@
 
 public class MapProcessorTest {
 
-  private final Pipe<String> input = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
-  private final Pipe<String> output = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+  private final Pipe<String> input =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "intput_pipe");
+  private final Pipe<String> output =
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "output_pipe");
 
   @SuppressWarnings("unchecked")
   private final Function<String, String> processor = mock(Function.class);

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/PipeTest.java

@@ -30,7 +30,8 @@ public class PipeTest {
   private final Counter inputCounter = mock(Counter.class);
   private final Counter outputCounter = mock(Counter.class);
   private final Counter abortedItemCounter = mock(Counter.class);
-  private final Pipe<String> pipe = new Pipe<>(5, inputCounter, outputCounter, abortedItemCounter);
+  private final Pipe<String> pipe =
+      new Pipe<>(5, inputCounter, outputCounter, abortedItemCounter, "test_pipe");
 
   @Test
   public void shouldNotHaveMoreWhenEmptyAndClosed() {

services/pipeline/src/test/java/org/hyperledger/besu/services/pipeline/ProcessingStageTest.java

@@ -34,9 +34,9 @@
 public class ProcessingStageTest {
 
   private final Pipe<String> inputPipe =
-      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "input_pipe");
   private final Pipe<String> outputPipe =
-      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER);
+      new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER, NO_OP_COUNTER, "output_pipe");
   @Mock private Processor<String, String> singleStep;
   private ProcessingStage<String, String> stage;