@VisibleForTesting
  static Map<String, Object> forCodec(
      Map<String, Object> codec, boolean replaceWithByteArrayCoder) {
    String coderType = (String) codec.get(PropertyNames.OBJECT_TYPE_NAME);

    // Handle well known coders.
    if (LENGTH_PREFIX_CODER_TYPE.equals(coderType)) {
      if (replaceWithByteArrayCoder) {
        return CloudObjects.asCloudObject(
            LENGTH_PREFIXED_BYTE_ARRAY_CODER, /*sdkComponents=*/ null);
      }
      return codec;
    } else if (WELL_KNOWN_CODER_TYPES.contains(coderType)) {
      // The runner knows about these types and can instantiate them so handle their
      // component encodings.
      Map<String, Object> prefixedCodec = new HashMap<>(codec);

      // Recursively replace component encodings
      if (codec.containsKey(PropertyNames.COMPONENT_ENCODINGS)) {
        List<Map<String, Object>> prefixedComponents = new ArrayList<>();
        for (Map<String, Object> component :
            (Iterable<Map<String, Object>>) codec.get(PropertyNames.COMPONENT_ENCODINGS)) {
          prefixedComponents.add(forCodec(component, replaceWithByteArrayCoder));
        }
        prefixedCodec.put(PropertyNames.COMPONENT_ENCODINGS, prefixedComponents);
      }
      return prefixedCodec;
    }

    // Wrap unknown coders with length prefix coder.
    if (replaceWithByteArrayCoder) {
      return CloudObjects.asCloudObject(LENGTH_PREFIXED_BYTE_ARRAY_CODER, /*sdkComponents=*/ null);
    } else {
      Map<String, Object> prefixedCodec = new HashMap<>();
      prefixedCodec.put(PropertyNames.OBJECT_TYPE_NAME, LENGTH_PREFIX_CODER_TYPE);
      prefixedCodec.put(PropertyNames.COMPONENT_ENCODINGS, ImmutableList.of(codec));
      return prefixedCodec;
    }
  }
@Test
  public void testLengthPrefixAndReplaceUnknownCoder() throws Exception {
    Coder<WindowedValue<KV<String, Integer>>> windowedValueCoder =
        WindowedValue.getFullCoder(
            KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()), GlobalWindow.Coder.INSTANCE);

    Map<String, Object> lengthPrefixedCoderCloudObject =
        forCodec(CloudObjects.asCloudObject(windowedValueCoder, /*sdkComponents=*/ null), true);

    assertEqualsAsJson(
        CloudObjects.asCloudObject(prefixedAndReplacedWindowedValueCoder, /*sdkComponents=*/ null),
        lengthPrefixedCoderCloudObject);
  }