public <OutT> void registerInputMessageStreams(
      PValue pvalue, List<? extends InputDescriptor<KV<?, OpMessage<OutT>>, ?>> inputDescriptors) {
    registerInputMessageStreams(pvalue, inputDescriptors, this::registerMessageStream);
  }
@Test
  public void testRegisterInputMessageStreams() {
    final PCollection output = mock(PCollection.class);
    List<String> topics = Arrays.asList("stream1", "stream2");
    List inputDescriptors =
        topics.stream()
            .map(topicName -> createSamzaInputDescriptor(topicName, topicName))
            .collect(Collectors.toList());

    translationContext.registerInputMessageStreams(output, inputDescriptors);

    assertNotNull(translationContext.getMessageStream(output));
  }