Java 类com.google.common.base.Supplier 实例源码

/**
 * Wait for the datanodes in the cluster to process any block
 * deletions that have already been asynchronously queued.
 */
public static void waitForDNDeletions(final MiniDFSCluster cluster)
    throws TimeoutException, InterruptedException {
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      for (DataNode dn : cluster.getDataNodes()) {
        if (DataNodeTestUtils.getPendingAsyncDeletions(dn) > 0) {
          return false;
        }
      }
      return true;
    }
  }, 1000, 10000);

}

private void waitForBlockReport(
    final DatanodeProtocolClientSideTranslatorPB mockNN1,
    final DatanodeProtocolClientSideTranslatorPB mockNN2)
        throws Exception {
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      return get(mockNN1) || get(mockNN2);
    }

    private Boolean get(DatanodeProtocolClientSideTranslatorPB mockNN) {
      try {
        Mockito.verify(mockNN).blockReport(
                Mockito.<DatanodeRegistration>anyObject(),
                Mockito.eq(FAKE_BPID),
                Mockito.<StorageBlockReport[]>anyObject(),
                Mockito.<BlockReportContext>anyObject());
        return true;
      } catch (Throwable t) {
        LOG.info("waiting on block report: " + t.getMessage());
        return false;
      }
    }
  }, 500, 10000);
}

private Supplier<Map<String, WorkingSetManager>> initContributions() {

        return memoize(() -> {

            if (!isRunning()) {
                return emptyMap();
            }

            final Builder<String, WorkingSetManager> builder = ImmutableMap.builder();
            final IConfigurationElement[] elements = getExtensionRegistry()
                    .getConfigurationElementsFor(EXTENSION_POINT_ID);
            for (final IConfigurationElement element : elements) {
                try {
                    final WorkingSetManager manager = (WorkingSetManager) element
                            .createExecutableExtension(CLASS_ATTRIBUTE);
                    injector.injectMembers(manager);
                    builder.put(manager.getId(), manager);
                } catch (final CoreException e) {
                    LOGGER.error("Error while trying to instantiate working set manager.", e);
                }
            }

            return builder.build();
        });
    }

/**
 * Wraps the given callable such that for the duration of {@link Callable#call} the thread that is
 * running will have the given name.
 *
 *
 * @param callable The callable to wrap
 * @param nameSupplier The supplier of thread names, {@link Supplier#get get} will be called once
 *     for each invocation of the wrapped callable.
 */
@GwtIncompatible // threads
static <T> Callable<T> threadRenaming(
    final Callable<T> callable, final Supplier<String> nameSupplier) {
  checkNotNull(nameSupplier);
  checkNotNull(callable);
  return new Callable<T>() {
    @Override
    public T call() throws Exception {
      Thread currentThread = Thread.currentThread();
      String oldName = currentThread.getName();
      boolean restoreName = trySetName(nameSupplier.get(), currentThread);
      try {
        return callable.call();
      } finally {
        if (restoreName) {
          boolean unused = trySetName(oldName, currentThread);
        }
      }
    }
  };
}

/**
 * Creates a new gauge metric.
 *
 * <p>Note that the order of the labels is significant.
 */
@Override
@CanIgnoreReturnValue
public <V> Metric<V> newGauge(
    String name,
    String description,
    String valueDisplayName,
    ImmutableSet<LabelDescriptor> labels,
    Supplier<ImmutableMap<ImmutableList<String>, V>> metricCallback,
    Class<V> valueClass) {
  VirtualMetric<V> metric =
      new VirtualMetric<>(
          name, description, valueDisplayName, labels, metricCallback, valueClass);
  registerMetric(name, metric);
  logger.info("Registered new callback metric: " + name);

  return metric;
}

/**
 * Save an asset && create blob.
 *
 * @return blob content
 */
static Content saveAsset(final StorageTx tx,
                         final Asset asset,
                         final Supplier<InputStream> contentSupplier,
                         final String contentType,
                         @Nullable final AttributesMap contentAttributes) throws IOException
{
  Content.applyToAsset(asset, Content.maintainLastModified(asset, contentAttributes));
  AssetBlob assetBlob = tx.setBlob(
      asset, asset.name(), contentSupplier, HASH_ALGORITHMS, null, contentType, false
  );

  asset.markAsDownloaded();
  tx.saveAsset(asset);
  return toContent(asset, assetBlob.getBlob());
}

/**
 * Ensure that {@link ReconfigurableBase#startReconfigurationTask} unsets the
 * property in its parent's configuration when the new value is null.
 * @throws IOException
 */
@Test (timeout=300000)
public void testConfIsUnsetAsync() throws ReconfigurationException,
    IOException, TimeoutException, InterruptedException {
  final String property = "FOO";
  final String value1 = "value1";

  final Configuration conf = new Configuration();
  conf.set(property, value1);
  final Configuration newConf = new Configuration();

  final ReconfigurableBase reconfigurable = makeReconfigurable(
      conf, newConf, Arrays.asList(property));

  // Kick off a reconfiguration task and wait until it completes.
  reconfigurable.startReconfigurationTask();
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      return reconfigurable.getReconfigurationTaskStatus().stopped();
    }
  }, 100, 60000);
  assertNull(reconfigurable.getConf().get(property));
}

private void waitForBlockReport(final DatanodeProtocolClientSideTranslatorPB mockNN)
    throws Exception {
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      try {
        Mockito.verify(mockNN).blockReport(
            Mockito.<DatanodeRegistration>anyObject(),  
            Mockito.eq(FAKE_BPID),
            Mockito.<StorageBlockReport[]>anyObject(),
            Mockito.<BlockReportContext>anyObject());
        return true;
      } catch (Throwable t) {
        LOG.info("waiting on block report: " + t.getMessage());
        return false;
      }
    }
  }, 500, 10000);
}

@Override
public FieldAccessTarget getTarget() {
    Supplier<Optional<JavaField>> fieldSupplier = new Supplier<Optional<JavaField>>() {
        @Override
        public Optional<JavaField> get() {
            return uniqueTargetIn(tryFindMatchingTargets(targetOwner.getAllFields(), record.target));
        }
    };
    JavaClass fieldType = classes.getOrResolve(Type.getType(record.target.desc).getClassName());
    return new FieldAccessTargetBuilder()
            .withOwner(targetOwner)
            .withName(record.target.name)
            .withType(fieldType)
            .withField(fieldSupplier)
            .build();
}

private ReceivedDeletedBlockInfo[] waitForBlockReceived(
    final ExtendedBlock fakeBlock,
    final DatanodeProtocolClientSideTranslatorPB mockNN) throws Exception {
  final String fakeBlockPoolId = fakeBlock.getBlockPoolId();
  final ArgumentCaptor<StorageReceivedDeletedBlocks[]> captor =
    ArgumentCaptor.forClass(StorageReceivedDeletedBlocks[].class);
  GenericTestUtils.waitFor(new Supplier<Boolean>() {

    @Override
    public Boolean get() {
      try {
        Mockito.verify(mockNN).blockReceivedAndDeleted(
          Mockito.<DatanodeRegistration>anyObject(),
          Mockito.eq(fakeBlockPoolId),
          captor.capture());
        return true;
      } catch (Throwable t) {
        return false;
      }
    }
  }, 100, 10000);
  return captor.getValue()[0].getBlocks();
}

/**
 * Returns a {@link Collector} that accumulates elements into a {@code Table} created using the
 * specified supplier, whose cells are generated by applying the provided mapping functions to the
 * input elements. Cells are inserted into the generated {@code Table} in encounter order.
 *
 * <p>If multiple input elements map to the same row and column, an {@code IllegalStateException}
 * is thrown when the collection operation is performed.
 *
 * @since 21.0
 */
@Beta
public static <T, R, C, V, I extends Table<R, C, V>> Collector<T, ?, I> toTable(
    java.util.function.Function<? super T, ? extends R> rowFunction,
    java.util.function.Function<? super T, ? extends C> columnFunction,
    java.util.function.Function<? super T, ? extends V> valueFunction,
    java.util.function.Supplier<I> tableSupplier) {
  return toTable(
      rowFunction,
      columnFunction,
      valueFunction,
      (v1, v2) -> {
        throw new IllegalStateException("Conflicting values " + v1 + " and " + v2);
      },
      tableSupplier);
}

/**
 * Wraps the given callable such that for the duration of {@link Callable#call} the thread that is
 * running will have the given name.
 *
 *
 * @param callable The callable to wrap
 * @param nameSupplier The supplier of thread names, {@link Supplier#get get} will be called once
 *     for each invocation of the wrapped callable.
 */
@GwtIncompatible // threads
static <T> Callable<T> threadRenaming(
    final Callable<T> callable, final Supplier<String> nameSupplier) {
  checkNotNull(nameSupplier);
  checkNotNull(callable);
  return new Callable<T>() {
    @Override
    public T call() throws Exception {
      Thread currentThread = Thread.currentThread();
      String oldName = currentThread.getName();
      boolean restoreName = trySetName(nameSupplier.get(), currentThread);
      try {
        return callable.call();
      } finally {
        if (restoreName) {
          boolean unused = trySetName(oldName, currentThread);
        }
      }
    }
  };
}

/**
 * Wraps the given runnable such that for the duration of {@link Runnable#run} the thread that is
 * running with have the given name.
 *
 *
 * @param task The Runnable to wrap
 * @param nameSupplier The supplier of thread names, {@link Supplier#get get} will be called once
 *     for each invocation of the wrapped callable.
 */
@GwtIncompatible // threads
static Runnable threadRenaming(final Runnable task, final Supplier<String> nameSupplier) {
  checkNotNull(nameSupplier);
  checkNotNull(task);
  return new Runnable() {
    @Override
    public void run() {
      Thread currentThread = Thread.currentThread();
      String oldName = currentThread.getName();
      boolean restoreName = trySetName(nameSupplier.get(), currentThread);
      try {
        task.run();
      } finally {
        if (restoreName) {
          boolean unused = trySetName(oldName, currentThread);
        }
      }
    }
  };
}

/**
 * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run in.
 *
 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
 * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
 * prevents the renaming then it will be skipped but the tasks will still execute.
 *
 *
 * @param executor The executor to decorate
 * @param nameSupplier The source of names for each task
 */
@GwtIncompatible // concurrency
static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) {
  checkNotNull(executor);
  checkNotNull(nameSupplier);
  if (isAppEngine()) {
    // AppEngine doesn't support thread renaming, so don't even try
    return executor;
  }
  return new Executor() {
    @Override
    public void execute(Runnable command) {
      executor.execute(Callables.threadRenaming(command, nameSupplier));
    }
  };
}

public static void waitFor(Supplier<Boolean> check,
    int checkEveryMillis, int waitForMillis)
    throws TimeoutException, InterruptedException
{
  long st = Time.now();
  do {
    boolean result = check.get();
    if (result) {
      return;
    }

    Thread.sleep(checkEveryMillis);
  } while (Time.now() - st < waitForMillis);

  throw new TimeoutException("Timed out waiting for condition. " +
      "Thread diagnostics:\n" +
      TimedOutTestsListener.buildThreadDiagnosticString());
}

@Override protected Table<String, Integer, Character> create(
    Object... data) {
  Supplier<TreeMap<Integer, Character>> factory
      = new Supplier<TreeMap<Integer, Character>>() {
        @Override
        public TreeMap<Integer, Character> get() {
          return Maps.newTreeMap();
        }
      };
  Map<String, Map<Integer, Character>> backingMap
      = Maps.newLinkedHashMap();
  Table<String, Integer, Character> table
      = Tables.newCustomTable(backingMap, factory);
  populate(table, data);
  return table;
}

@GwtIncompatible // threads
public void testRenaming_exceptionalReturn() throws Exception {
  String oldName = Thread.currentThread().getName();
  final Supplier<String> newName = Suppliers.ofInstance("MyCrazyThreadName");
  class MyException extends Exception {}
  Callable<Void> callable = new Callable<Void>() {
    @Override public Void call() throws Exception {
      assertEquals(Thread.currentThread().getName(), newName.get());
      throw new MyException();
    }
  };
  try {
    Callables.threadRenaming(callable, newName).call();
    fail();
  } catch (MyException expected) {}
  assertEquals(oldName, Thread.currentThread().getName());
}

@Autowired
VacuumToolApplication(
    @Value("#{replicaHiveConf}") HiveConf conf,
    @Value("#{replicaMetaStoreClientSupplier}") Supplier<CloseableMetaStoreClient> clientSupplier,
    LegacyReplicaPathRepository legacyReplicaPathRepository,
    HousekeepingService housekeepingService,
    TableReplications replications,
    @Value("${dry-run:false}") boolean isDryRun,
    @Value("${partition-batch-size:1000}") short batchSize,
    @Value("${expected-path-count:10000}") int expectedPathCount) {
  this.conf = conf;
  this.clientSupplier = clientSupplier;
  this.legacyReplicaPathRepository = legacyReplicaPathRepository;
  this.housekeepingService = housekeepingService;
  this.isDryRun = isDryRun;
  this.batchSize = batchSize;
  this.expectedPathCount = expectedPathCount;
  tableReplications = replications.getTableReplications();
  vacuumEventId = "vacuum-" + DateTime.now(DateTimeZone.UTC);
}

/**
 * Wraps the given runnable such that for the duration of {@link Runnable#run} the thread that is
 * running with have the given name.
 *
 *
 * @param task The Runnable to wrap
 * @param nameSupplier The supplier of thread names, {@link Supplier#get get} will be called once
 *     for each invocation of the wrapped callable.
 */
@GwtIncompatible // threads
static Runnable threadRenaming(final Runnable task, final Supplier<String> nameSupplier) {
  checkNotNull(nameSupplier);
  checkNotNull(task);
  return new Runnable() {
    @Override
    public void run() {
      Thread currentThread = Thread.currentThread();
      String oldName = currentThread.getName();
      boolean restoreName = trySetName(nameSupplier.get(), currentThread);
      try {
        task.run();
      } finally {
        if (restoreName) {
          boolean unused = trySetName(oldName, currentThread);
        }
      }
    }
  };
}

void completeMembers(final ImportContext context) {
    fields = context.createFields(this);
    methods = context.createMethods(this);
    constructors = context.createConstructors(this);
    staticInitializer = context.createStaticInitializer(this);
    codeUnits = ImmutableSet.<JavaCodeUnit>builder()
            .addAll(methods).addAll(constructors).addAll(staticInitializer.asSet())
            .build();
    members = ImmutableSet.<JavaMember>builder()
            .addAll(fields)
            .addAll(methods)
            .addAll(constructors)
            .build();
    this.annotations = Suppliers.memoize(new Supplier<Map<String, JavaAnnotation>>() {
        @Override
        public Map<String, JavaAnnotation> get() {
            return context.createAnnotations(JavaClass.this);
        }
    });
}

static <T extends Completable> T getInstance(Class<T> type, Supplier<T> supplier) {
  ClassToInstanceMap<Completable> components = state().components;
  @Nullable T instance = components.getInstance(type);
  if (instance == null) {
    instance = supplier.get();
    components.putInstance(type, instance);
  }
  return instance;
}

/**
 * Test for the case where the client beings to read a long block, but doesn't
 * read bytes off the stream quickly. The datanode should time out sending the
 * chunks and the transceiver should die, even if it has a long keepalive.
 */
@Test(timeout=300000)
public void testSlowReader() throws Exception {
  // Set a client socket cache expiry time much longer than 
  // the datanode-side expiration time.
  final long CLIENT_EXPIRY_MS = 600000L;
  Configuration clientConf = new Configuration(conf);
  clientConf.setLong(DFS_CLIENT_SOCKET_CACHE_EXPIRY_MSEC_KEY, CLIENT_EXPIRY_MS);
  clientConf.set(DFS_CLIENT_CONTEXT, "testSlowReader");
  DistributedFileSystem fs =
      (DistributedFileSystem)FileSystem.get(cluster.getURI(),
          clientConf);
  // Restart the DN with a shorter write timeout.
  DataNodeProperties props = cluster.stopDataNode(0);
  props.conf.setInt(DFS_DATANODE_SOCKET_WRITE_TIMEOUT_KEY,
      WRITE_TIMEOUT);
  props.conf.setInt(DFS_DATANODE_SOCKET_REUSE_KEEPALIVE_KEY,
      120000);
  assertTrue(cluster.restartDataNode(props, true));
  dn = cluster.getDataNodes().get(0);
  // Wait for heartbeats to avoid a startup race where we
  // try to write the block while the DN is still starting.
  cluster.triggerHeartbeats();

  DFSTestUtil.createFile(fs, TEST_FILE, 1024*1024*8L, (short)1, 0L);
  FSDataInputStream stm = fs.open(TEST_FILE);
  stm.read();
  assertXceiverCount(1);

  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      // DN should time out in sendChunks, and this should force
      // the xceiver to exit.
      return getXceiverCountWithoutServer() == 0;
    }
  }, 500, 50000);

  IOUtils.closeStream(stm);
}

private void waitForInitialization(final BPOfferService bpos)
    throws Exception {
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      return bpos.isAlive() && bpos.isInitialized();
    }
  }, 100, 10000);
}

public NavMeshSegmentedAStarPathPlanner(Supplier<Collection<NavPoint>> navGraphProvider, NavMeshDropGrounder grounder, NavMesh navMesh, Logger log) {
    this.navGraphProvider = navGraphProvider;
    this.grounder = grounder;
    this.navMesh = navMesh;
    this.log = log;

    this.fwMap = null;
    this.allNavPoints = null;
    this.hasTeleports = false;
}

@Inject
public BlobShardSizeExpression(final BlobShard blobShard) {
    totalUsageSupplier = Suppliers.memoizeWithExpiration(new Supplier<Long>() {
        @Override
        public Long get() {
            return blobShard.blobStats().totalUsage();
        }
    }, 10, TimeUnit.SECONDS);
}

/**
 * Test that, once the queue eclipses the configure size limit,
 * calls to journal more data are rejected.
 */
@Test
public void testQueueLimiting() throws Exception {
  // Block the underlying fake proxy from actually completing any calls.
  DelayAnswer delayer = new DelayAnswer(LOG);
  Mockito.doAnswer(delayer).when(mockProxy).journal(
      Mockito.<RequestInfo>any(),
      Mockito.eq(1L), Mockito.eq(1L),
      Mockito.eq(1), Mockito.same(FAKE_DATA));

  // Queue up the maximum number of calls.
  int numToQueue = LIMIT_QUEUE_SIZE_BYTES / FAKE_DATA.length;
  for (int i = 1; i <= numToQueue; i++) {
    ch.sendEdits(1L, (long)i, 1, FAKE_DATA);
  }

  // The accounting should show the correct total number queued.
  assertEquals(LIMIT_QUEUE_SIZE_BYTES, ch.getQueuedEditsSize());

  // Trying to queue any more should fail.
  try {
    ch.sendEdits(1L, numToQueue + 1, 1, FAKE_DATA).get(1, TimeUnit.SECONDS);
    fail("Did not fail to queue more calls after queue was full");
  } catch (ExecutionException ee) {
    if (!(ee.getCause() instanceof LoggerTooFarBehindException)) {
      throw ee;
    }
  }

  delayer.proceed();

  // After we allow it to proceeed, it should chug through the original queue
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      return ch.getQueuedEditsSize() == 0;
    }
  }, 10, 1000);
}

@GwtIncompatible // java.io.ObjectInputStream
@SuppressWarnings("unchecked") // reading data stored by writeObject
private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
  stream.defaultReadObject();
  factory = (Supplier<? extends Set<V>>) stream.readObject();
  Map<K, Collection<V>> map = (Map<K, Collection<V>>) stream.readObject();
  setMap(map);
}

public void testGenericInterface() {
  // test the 1st generic interface on the class
  Type fType = Supplier.class.getTypeParameters()[0];
  assertEquals(Integer.class,
      TypeToken.of(IntegerStringFunction.class).resolveType(fType)
          .getRawType());

  // test the 2nd generic interface on the class
  Type predicateParameterType = Predicate.class.getTypeParameters()[0];
  assertEquals(new TypeToken<List<String>>() {}.getType(),
      TypeToken.of(IntegerStringFunction.class).resolveType(predicateParameterType)
          .getType());
}

private static void waitForLogRollInSharedDir(MiniDFSCluster cluster,
    long startTxId) throws Exception {
  URI sharedUri = cluster.getSharedEditsDir(0, 1);
  File sharedDir = new File(sharedUri.getPath(), "current");
  final File expectedLog = new File(sharedDir,
      NNStorage.getInProgressEditsFileName(startTxId));

  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    @Override
    public Boolean get() {
      return expectedLog.exists();
    }
  }, 100, 10000);
}

@GwtIncompatible // java.io.ObjectInputStream
@SuppressWarnings("unchecked") // reading data stored by writeObject
private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
  stream.defaultReadObject();
  factory = (Supplier<? extends SortedSet<V>>) stream.readObject();
  valueComparator = factory.get().comparator();
  Map<K, Collection<V>> map = (Map<K, Collection<V>>) stream.readObject();
  setMap(map);
}

/**
 * Uses an {@link EnumSet} to store value collections.
 */
public <V0 extends Enum<V0>> SetMultimapBuilder<K0, V0> enumSetValues(
    final Class<V0> valueClass) {
  checkNotNull(valueClass, "valueClass");
  return new SetMultimapBuilder<K0, V0>() {
    @Override
    public <K extends K0, V extends V0> SetMultimap<K, V> build() {
      // V must actually be V0, since enums are effectively final
      // (their subclasses are inaccessible)
      @SuppressWarnings({"unchecked", "rawtypes"})
      Supplier<Set<V>> factory = (Supplier) new EnumSetSupplier<V0>(valueClass);
      return Multimaps.newSetMultimap(MultimapBuilderWithKeys.this.<K, V>createMap(), factory);
    }
  };
}

private CompactStriped(int stripes, Supplier<L> supplier) {
  super(stripes);
  Preconditions.checkArgument(stripes <= Ints.MAX_POWER_OF_TWO, "Stripes must be <= 2^30)");

  this.array = new Object[mask + 1];
  for (int i = 0; i < array.length; i++) {
    array[i] = supplier.get();
  }
}

@Override
protected final void doStart() {
  executorService =
      MoreExecutors.renamingDecorator(
          executor(),
          new Supplier<String>() {
            @Override
            public String get() {
              return serviceName() + " " + state();
            }
          });
  executorService.execute(
      new Runnable() {
        @Override
        public void run() {
          lock.lock();
          try {
            startUp();
            runningTask = scheduler().schedule(delegate, executorService, task);
            notifyStarted();
          } catch (Throwable t) {
            notifyFailed(t);
            if (runningTask != null) {
              // prevent the task from running if possible
              runningTask.cancel(false);
            }
          } finally {
            lock.unlock();
          }
        }
      });
}

/**
 * Returns a {@link Collector} that accumulates elements into a {@code Table} created using the
 * specified supplier, whose cells are generated by applying the provided mapping functions to the
 * input elements. Cells are inserted into the generated {@code Table} in encounter order.
 *
 * <p>If multiple input elements map to the same row and column, the specified merging function is
 * used to combine the values. Like {@link
 * java.util.stream.Collectors#toMap(java.util.function.Function, java.util.function.Function,
 * BinaryOperator, java.util.function.Supplier)}, this Collector throws a {@code
 * NullPointerException} on null values returned from {@code valueFunction}, and treats nulls
 * returned from {@code mergeFunction} as removals of that row/column pair.
 *
 * @since 21.0
 */
public static <T, R, C, V, I extends Table<R, C, V>> Collector<T, ?, I> toTable(
    java.util.function.Function<? super T, ? extends R> rowFunction,
    java.util.function.Function<? super T, ? extends C> columnFunction,
    java.util.function.Function<? super T, ? extends V> valueFunction,
    BinaryOperator<V> mergeFunction,
    java.util.function.Supplier<I> tableSupplier) {
  checkNotNull(rowFunction);
  checkNotNull(columnFunction);
  checkNotNull(valueFunction);
  checkNotNull(mergeFunction);
  checkNotNull(tableSupplier);
  return Collector.of(
      tableSupplier,
      (table, input) ->
          merge(
              table,
              rowFunction.apply(input),
              columnFunction.apply(input),
              valueFunction.apply(input),
              mergeFunction),
      (table1, table2) -> {
        for (Table.Cell<R, C, V> cell2 : table2.cellSet()) {
          merge(table1, cell2.getRowKey(), cell2.getColumnKey(), cell2.getValue(), mergeFunction);
        }
        return table1;
      });
}

/**
 * Creates a {@code Striped<Lock>} with eagerly initialized, strongly referenced locks. Every lock
 * is reentrant.
 *
 * @param stripes the minimum number of stripes (locks) required
 * @return a new {@code Striped<Lock>}
 */
public static Striped<Lock> lock(int stripes) {
  return new CompactStriped<Lock>(
      stripes,
      new Supplier<Lock>() {
        @Override
        public Lock get() {
          return new PaddedLock();
        }
      });
}

@Inject
public ShardSizeExpression(final IndexShard indexShard) {
    sizeSupplier = Suppliers.memoizeWithExpiration(new Supplier<Long>() {
        @Override
        public Long get() {
            return indexShard.storeStats().getSizeInBytes();
        }
    }, 10, TimeUnit.SECONDS);
}

void check() {
  runTester();
  FromTo<?, ?> defaultFunction = (FromTo<?, ?>) getDefaultParameterValue(0);
  assertEquals(0, defaultFunction.apply(null));
  Supplier<?> defaultSupplier = (Supplier<?>) defaultFunction;
  assertEquals(Long.valueOf(0), defaultSupplier.get());
}

protected CommandProcess(Command command, RawProcess rawProcess) throws CommandStartException {
  this.command = checkNotNull(command);
  this.rawProcess = checkNotNull(rawProcess);
  stdoutStream = new CapturingOutputStream();
  stderrStream = new CapturingOutputStream();

  Supplier<Level> ioLogLevel = new Supplier<Level>() {
    @Override
    public Level get() {
      return CommandProcess.this.rawProcess.isAlive() ? Level.WARNING : Level.FINE;
    }
  };

  try {
    stdinPump =
        maybeStartCopyFromSourceToPipe(command.stdinSource(), rawProcess.stdinPipe(), ioLogLevel);
    stdoutPump =
        maybeStartCopyFromPipeToSink(
            rawProcess.stdoutPipe(),
            command.stdoutSink(),
            stdoutStream,
            stderrStream,
            ioLogLevel);
    stderrPump =
        maybeStartCopyFromPipeToSink(
            rawProcess.stderrPipe(),
            command.stderrSink(),
            stdoutStream,
            stderrStream,
            ioLogLevel);
  } catch (IOException e) {
    throw new CommandStartException(command, e);
  }
}

public void testGenericInterface() {
  // test the 1st generic interface on the class
  Type fType = Supplier.class.getTypeParameters()[0];
  assertEquals(Integer.class,
      TypeToken.of(IntegerStringFunction.class).resolveType(fType)
          .getRawType());

  // test the 2nd generic interface on the class
  Type predicateParameterType = Predicate.class.getTypeParameters()[0];
  assertEquals(new TypeToken<List<String>>() {}.getType(),
      TypeToken.of(IntegerStringFunction.class).resolveType(predicateParameterType)
          .getType());
}

public Projector visitSourceIndexWriterProjection(SourceIndexWriterProjection projection, Context context) {
    ImplementationSymbolVisitor.Context symbolContext = new ImplementationSymbolVisitor.Context();
    List<Input<?>> partitionedByInputs = new ArrayList<>(projection.partitionedBySymbols().size());
    for (Symbol partitionedBySymbol : projection.partitionedBySymbols()) {
        partitionedByInputs.add(symbolVisitor.process(partitionedBySymbol, symbolContext));
    }
    Input<?> sourceInput = symbolVisitor.process(projection.rawSource(), symbolContext);
    Supplier<String> indexNameResolver =
            IndexNameResolver.create(projection.tableIdent(), projection.partitionIdent(), partitionedByInputs);
    return new IndexWriterProjector(
            clusterService,
            functions,
            indexNameExpressionResolver,
            clusterService.state().metaData().settings(),
            transportActionProvider,
            indexNameResolver,
            bulkRetryCoordinatorPool,
            projection.rawSourceReference(),
            projection.primaryKeys(),
            projection.ids(),
            projection.clusteredBy(),
            projection.clusteredByIdent(),
            sourceInput,
            symbolContext.collectExpressions(),
            projection.bulkActions(),
            projection.includes(),
            projection.excludes(),
            projection.autoCreateIndices(),
            projection.overwriteDuplicates(),
            context.jobId
    );
}