@VisibleForTesting
  static ParallelInstruction forParallelInstruction(
      ParallelInstruction input, boolean replaceWithByteArrayCoder) throws Exception {
    try {
      ParallelInstruction instruction = clone(input, ParallelInstruction.class);
      if (instruction.getRead() != null) {
        Source cloudSource = instruction.getRead().getSource();
        cloudSource.setCodec(forCodec(cloudSource.getCodec(), replaceWithByteArrayCoder));
      } else if (instruction.getWrite() != null) {
        com.google.api.services.dataflow.model.Sink cloudSink = instruction.getWrite().getSink();
        cloudSink.setCodec(forCodec(cloudSink.getCodec(), replaceWithByteArrayCoder));
      } else if (instruction.getParDo() != null) {
        instruction.setParDo(
            forParDoInstruction(instruction.getParDo(), replaceWithByteArrayCoder));
      } else if (instruction.getPartialGroupByKey() != null) {
        PartialGroupByKeyInstruction pgbk = instruction.getPartialGroupByKey();
        pgbk.setInputElementCodec(forCodec(pgbk.getInputElementCodec(), replaceWithByteArrayCoder));
      } else if (instruction.getFlatten() != null) {
        // FlattenInstructions have no codecs to wrap.
      } else {
        throw new RuntimeException("Unknown parallel instruction: " + input);
      }
      return instruction;
    } catch (IOException e) {
      throw new RuntimeException(
          String.format(
              "Failed to replace unknown coder with " + "LengthPrefixCoder for : {%s}", input),
          e);
    }
  }
@Test
  public void testLengthPrefixReadInstructionCoder() throws Exception {
    ReadInstruction readInstruction = new ReadInstruction();
    readInstruction.setSource(
        new Source()
            .setCodec(CloudObjects.asCloudObject(windowedValueCoder, /*sdkComponents=*/ null)));
    instruction.setRead(readInstruction);

    ParallelInstruction prefixedInstruction = forParallelInstruction(instruction, false);
    assertEqualsAsJson(
        CloudObjects.asCloudObject(prefixedWindowedValueCoder, /*sdkComponents=*/ null),
        prefixedInstruction.getRead().getSource().getCodec());
    // Should not mutate the instruction.
    assertEqualsAsJson(
        readInstruction.getSource().getCodec(),
        CloudObjects.asCloudObject(windowedValueCoder, /*sdkComponents=*/ null));
  }