public FnDataReceiver<WindowedValue<?>> getMultiplexingConsumer(String pCollectionId) {
    return pCollectionIdsToWrappedConsumer.computeIfAbsent(
        pCollectionId,
        pcId -> {
          if (!processBundleDescriptor.containsPcollections(pCollectionId)) {
            throw new IllegalArgumentException(
                String.format("Unknown PCollection id %s", pCollectionId));
          }
          String coderId =
              processBundleDescriptor.getPcollectionsOrThrow(pCollectionId).getCoderId();
          Coder<?> coder;
          OutputSampler<?> sampler = null;
          try {
            Coder<?> maybeWindowedValueInputCoder = rehydratedComponents.getCoder(coderId);

            if (dataSampler != null) {
              sampler = dataSampler.sampleOutput(pCollectionId, maybeWindowedValueInputCoder);
            }

            // TODO: Stop passing windowed value coders within PCollections.
            if (maybeWindowedValueInputCoder instanceof WindowedValue.WindowedValueCoder) {
              coder = ((WindowedValueCoder) maybeWindowedValueInputCoder).getValueCoder();
            } else {
              coder = maybeWindowedValueInputCoder;
            }
          } catch (IOException e) {
            throw new IllegalStateException(
                String.format("Unable to materialize coder %s", coderId), e);
          }
          List<ConsumerAndMetadata> consumerAndMetadatas =
              pCollectionIdsToConsumers.computeIfAbsent(
                  pCollectionId, (unused) -> new ArrayList<>());

          if (consumerAndMetadatas.size() == 1) {
            ConsumerAndMetadata consumerAndMetadata = consumerAndMetadatas.get(0);
            if (consumerAndMetadata.getConsumer() instanceof HandlesSplits) {
              return new SplittingMetricTrackingFnDataReceiver(
                  pcId, coder, consumerAndMetadata, sampler);
            }
            return new MetricTrackingFnDataReceiver(pcId, coder, consumerAndMetadata, sampler);
          } else {
            /* TODO(SDF), Consider supporting splitting each consumer individually. This would never
            come up in the existing SDF expansion, but might be useful to support fused SDF nodes.
            This would require dedicated delivery of the split results to each of the consumers
            separately. */
            return new MultiplexingMetricTrackingFnDataReceiver(
                pcId, coder, consumerAndMetadatas, sampler);
          }
        });
  }
@Test
  public void noConsumers() throws Exception {
    ShortIdMap shortIds = new ShortIdMap();
    BundleProgressReporter.InMemory reporterAndRegistrar = new BundleProgressReporter.InMemory();
    PCollectionConsumerRegistry consumers =
        new PCollectionConsumerRegistry(
            sampler.create(), shortIds, reporterAndRegistrar, TEST_DESCRIPTOR);

    FnDataReceiver<WindowedValue<String>> wrapperConsumer =
        (FnDataReceiver<WindowedValue<String>>)
            (FnDataReceiver) consumers.getMultiplexingConsumer(P_COLLECTION_A);
    String elementValue = "elem";
    WindowedValue<String> element = valueInGlobalWindow(elementValue);
    int numElements = 10;
    for (int i = 0; i < numElements; i++) {
      wrapperConsumer.accept(element);
    }

    List<MonitoringInfo> expected = new ArrayList<>();

    SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder();
    builder.setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT);
    builder.setLabel(MonitoringInfoConstants.Labels.PCOLLECTION, P_COLLECTION_A);
    builder.setInt64SumValue(numElements);
    expected.add(builder.build());

    long elementByteSize = StringUtf8Coder.of().getEncodedElementByteSize(elementValue);
    builder = new SimpleMonitoringInfoBuilder();
    builder.setUrn(Urns.SAMPLED_BYTE_SIZE);
    builder.setLabel(MonitoringInfoConstants.Labels.PCOLLECTION, P_COLLECTION_A);
    builder.setInt64DistributionValue(
        DistributionData.create(
            numElements * elementByteSize, numElements, elementByteSize, elementByteSize));
    expected.add(builder.build());

    Map<String, ByteString> actualData = new HashMap<>();
    reporterAndRegistrar.updateFinalMonitoringData(actualData);

    // Clear the timestamp before comparison.
    Iterable<MonitoringInfo> result =
        Iterables.filter(
            shortIds.toMonitoringInfo(actualData),
            monitoringInfo -> monitoringInfo.containsLabels(Labels.PCOLLECTION));

    assertThat(result, containsInAnyOrder(expected.toArray()));
  }