public static <T> WindowedValue.WindowedValueCoder<T> instantiateWindowedCoder(
      String collectionId, RunnerApi.Components components) {
    PipelineNode.PCollectionNode collectionNode =
        PipelineNode.pCollection(collectionId, components.getPcollectionsOrThrow(collectionId));
    try {
      return (WindowedValue.WindowedValueCoder<T>)
          WireCoders.<T>instantiateRunnerWireCoder(collectionNode, components);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
@Test
  public void testInstantiateWindowedCoder() throws IOException {
    Coder<KV<Long, String>> expectedValueCoder =
        KvCoder.of(VarLongCoder.of(), StringUtf8Coder.of());
    SdkComponents components = SdkComponents.create();
    components.registerEnvironment(Environments.createDockerEnvironment("java"));
    String collectionId =
        components.registerPCollection(
            PCollection.createPrimitiveOutputInternal(
                    Pipeline.create(),
                    WindowingStrategy.globalDefault(),
                    PCollection.IsBounded.BOUNDED,
                    expectedValueCoder)
                .setName("name"));

    assertEquals(
        expectedValueCoder,
        WindowUtils.instantiateWindowedCoder(collectionId, components.toComponents())
            .getValueCoder());
  }