content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.sdk.options;
  
  import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
  import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
  import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasNamespace;
  import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasType;
  import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasValue;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.allOf;
  import static org.hamcrest.Matchers.hasItem;
  import static org.hamcrest.Matchers.not;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assume.assumeFalse;
  
  import com.fasterxml.jackson.annotation.JsonIgnore;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.fasterxml.jackson.annotation.JsonValue;
  import com.fasterxml.jackson.core.JsonGenerator;
  import com.fasterxml.jackson.core.JsonParser;
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.DeserializationContext;
  import com.fasterxml.jackson.databind.JsonMappingException;
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.fasterxml.jackson.databind.SerializerProvider;
  import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
  import com.fasterxml.jackson.databind.annotation.JsonSerialize;
  import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
  import com.fasterxml.jackson.databind.ser.std.StdSerializer;
  import com.fasterxml.jackson.databind.util.TokenBuffer;
  import com.google.common.testing.EqualsTester;
  import java.io.IOException;
  import java.io.NotSerializableException;
  import java.io.Serializable;
  import java.lang.reflect.Method;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  import net.bytebuddy.ByteBuddy;
  import net.bytebuddy.description.modifier.Visibility;
  import net.bytebuddy.description.type.TypeDescription;
  import net.bytebuddy.dynamic.DynamicType;
  import org.apache.beam.sdk.testing.NeedsRunner;
  import org.apache.beam.sdk.testing.TestPipeline;
  import org.apache.beam.sdk.transforms.Create;
  import org.apache.beam.sdk.transforms.display.DisplayData;
  import org.apache.beam.sdk.util.SerializableUtils;
  import org.apache.beam.sdk.util.common.ReflectHelpers;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.ByteStreams;
  import org.apache.commons.lang3.SystemUtils;
  import org.checkerframework.checker.nullness.qual.Nullable;
  import org.hamcrest.Matchers;
  import org.joda.time.Instant;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.experimental.categories.Category;
  import org.junit.rules.ExpectedException;
  import org.junit.rules.ExternalResource;
  import org.junit.rules.TestRule;
  import org.junit.runner.RunWith;
  import org.junit.runners.JUnit4;
  
  /** Tests for {@link ProxyInvocationHandler}. */
  @RunWith(JUnit4.class)
  public class ProxyInvocationHandlerTest {
    @Rule public ExpectedException expectedException = ExpectedException.none();
  
    @Rule
    public TestRule resetPipelineOptionsRegistry =
        new ExternalResource() {
          @Override
          protected void before() {
            PipelineOptionsFactory.resetCache();
          }
        };
  
    private static final ObjectMapper MAPPER =
        new ObjectMapper()
            .registerModules(ObjectMapper.findModules(ReflectHelpers.findClassLoader()));
  
    /** A test interface with some primitives and objects. */
    public interface Simple extends PipelineOptions {
      boolean isOptionEnabled();
  
      void setOptionEnabled(boolean value);
  
      int getPrimitive();
  
      void setPrimitive(int value);
  
      String getString();
  
      void setString(String value);
    }
  
    @Rule public TestPipeline p = TestPipeline.create();
  
    @Test
    public void testPropertySettingAndGetting() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      Simple proxy = handler.as(Simple.class);
      proxy.setString("OBJECT");
      proxy.setOptionEnabled(true);
      proxy.setPrimitive(4);
      assertEquals("OBJECT", proxy.getString());
      assertTrue(proxy.isOptionEnabled());
      assertEquals(4, proxy.getPrimitive());
    }
  
    /** A test interface containing all the JLS default values. */
    public interface JLSDefaults extends PipelineOptions {
      boolean getBoolean();
  
      void setBoolean(boolean value);
  
      char getChar();
  
      void setChar(char value);
  
      byte getByte();
  
      void setByte(byte value);
  
      short getShort();
  
      void setShort(short value);
  
      int getInt();
  
      void setInt(int value);
  
      long getLong();
  
      void setLong(long value);
  
      float getFloat();
  
      void setFloat(float value);
  
      double getDouble();
  
      void setDouble(double value);
  
      Object getObject();
  
      void setObject(Object value);
    }
  
    @Test
    public void testGettingJLSDefaults() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      JLSDefaults proxy = handler.as(JLSDefaults.class);
      assertFalse(proxy.getBoolean());
      assertEquals('\0', proxy.getChar());
      assertEquals((byte) 0, proxy.getByte());
      assertEquals((short) 0, proxy.getShort());
      assertEquals(0, proxy.getInt());
      assertEquals(0L, proxy.getLong());
      assertEquals(0f, proxy.getFloat(), 0f);
      assertEquals(0d, proxy.getDouble(), 0d);
      assertNull(proxy.getObject());
    }
  
    /** A {@link DefaultValueFactory} that is used for testing. */
    public static class TestOptionFactory implements DefaultValueFactory<String> {
      @Override
      public String create(PipelineOptions options) {
        return "testOptionFactory";
      }
    }
  
    /** A test enum for testing {@link Default.Enum @Default.Enum}. */
    public enum EnumType {
      MyEnum("MyTestEnum");
  
      private final String value;
  
      EnumType(String value) {
        this.value = value;
      }
  
      @Override
      public String toString() {
        return value;
      }
    }
  
    /** A test interface containing all the {@link Default} annotations. */
    public interface DefaultAnnotations extends PipelineOptions {
      @Default.Boolean(true)
      boolean getBoolean();
  
      void setBoolean(boolean value);
  
      @Default.Character('a')
      char getChar();
  
      void setChar(char value);
  
      @Default.Byte((byte) 4)
      byte getByte();
  
      void setByte(byte value);
  
      @Default.Short((short) 5)
      short getShort();
  
      void setShort(short value);
  
      @Default.Integer(6)
      int getInt();
  
      void setInt(int value);
  
      @Default.Long(7L)
      long getLong();
  
      void setLong(long value);
  
      @Default.Float(8f)
      float getFloat();
  
      void setFloat(float value);
  
      @Default.Double(9d)
      double getDouble();
  
      void setDouble(double value);
  
      @Default.String("testString")
      String getString();
  
      void setString(String value);
  
      @Default.Class(DefaultAnnotations.class)
      Class<?> getClassOption();
  
      void setClassOption(Class<?> value);
  
      @Default.Enum("MyEnum")
      EnumType getEnum();
  
      void setEnum(EnumType value);
  
      @Default.InstanceFactory(TestOptionFactory.class)
      String getComplex();
  
      void setComplex(String value);
    }
  
    @Test
    public void testAnnotationDefaults() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      DefaultAnnotations proxy = handler.as(DefaultAnnotations.class);
      assertTrue(proxy.getBoolean());
      assertEquals('a', proxy.getChar());
      assertEquals((byte) 4, proxy.getByte());
      assertEquals((short) 5, proxy.getShort());
      assertEquals(6, proxy.getInt());
      assertEquals(7, proxy.getLong());
      assertEquals(8f, proxy.getFloat(), 0f);
      assertEquals(9d, proxy.getDouble(), 0d);
      assertEquals("testString", proxy.getString());
      assertEquals(DefaultAnnotations.class, proxy.getClassOption());
      assertEquals(EnumType.MyEnum, proxy.getEnum());
      assertEquals("testOptionFactory", proxy.getComplex());
    }
  
    @Test
    public void testEqualsAndHashCode() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      Simple proxy = handler.as(Simple.class);
      JLSDefaults sameAsProxy = proxy.as(JLSDefaults.class);
      ProxyInvocationHandler handler2 = new ProxyInvocationHandler(Maps.newHashMap());
      Simple proxy2 = handler2.as(Simple.class);
      JLSDefaults sameAsProxy2 = proxy2.as(JLSDefaults.class);
  
      new EqualsTester()
          .addEqualityGroup(handler, proxy, sameAsProxy)
          .addEqualityGroup(handler2, proxy2, sameAsProxy2)
          .testEquals();
    }
  
    /** A test interface for string with default. */
    public interface StringWithDefault extends PipelineOptions {
      @Default.String("testString")
      String getString();
  
      void setString(String value);
    }
  
    @Test
    public void testToString() throws Exception {
      // TODO: Java core test failing on windows, https://github.com/apache/beam/issues/20485
      assumeFalse(SystemUtils.IS_OS_WINDOWS);
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      StringWithDefault proxy = handler.as(StringWithDefault.class);
      proxy.setString("stringValue");
      DefaultAnnotations proxy2 = proxy.as(DefaultAnnotations.class);
      proxy2.setLong(57L);
      assertEquals(
          String.format("Current Settings:%n" + "  long: 57%n" + "  string: stringValue%n"),
          proxy.toString());
    }
  
    @Test
    public void testToStringAfterDeserializationContainsJsonEntries() throws Exception {
      // TODO: Java core test failing on windows, https://github.com/apache/beam/issues/20474
      assumeFalse(SystemUtils.IS_OS_WINDOWS);
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      StringWithDefault proxy = handler.as(StringWithDefault.class);
      Long optionsId = proxy.getOptionsId();
      proxy.setString("stringValue");
      DefaultAnnotations proxy2 = proxy.as(DefaultAnnotations.class);
      proxy2.setLong(57L);
      assertEquals(
          String.format(
              "Current Settings:%n"
                  + "  long: 57%n"
                  + "  optionsId: %d%n"
                  + "  string: \"stringValue\"%n",
              optionsId),
          serializeDeserialize(PipelineOptions.class, proxy2).toString());
    }
  
    @Test
    public void testToStringAfterDeserializationContainsOverriddenEntries() throws Exception {
      // TODO: Java core test failing on windows, https://github.com/apache/beam/issues/20473
      assumeFalse(SystemUtils.IS_OS_WINDOWS);
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      StringWithDefault proxy = handler.as(StringWithDefault.class);
      Long optionsId = proxy.getOptionsId();
      proxy.setString("stringValue");
      DefaultAnnotations proxy2 = proxy.as(DefaultAnnotations.class);
      proxy2.setLong(57L);
      Simple deserializedOptions = serializeDeserialize(Simple.class, proxy2);
      deserializedOptions.setString("overriddenValue");
      assertEquals(
          String.format(
              "Current Settings:%n"
                  + "  long: 57%n"
                  + "  optionsId: %d%n"
                  + "  string: overriddenValue%n",
              optionsId),
          deserializedOptions.toString());
    }
  
    /** A test interface containing an unknown method. */
    public interface UnknownMethod {
      void unknownMethod();
    }
  
    @Test
    public void testInvokeWithUnknownMethod() throws Exception {
      expectedException.expect(RuntimeException.class);
      expectedException.expectMessage(
          "Unknown method [public abstract void "
              + "org.apache.beam.sdk.options.ProxyInvocationHandlerTest$UnknownMethod.unknownMethod()] "
              + "invoked with args [null].");
  
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      handler.invoke(handler, UnknownMethod.class.getMethod("unknownMethod"), null);
    }
  
    /** A test interface that extends another interface. */
    public interface SubClass extends Simple {
      String getExtended();
  
      void setExtended(String value);
    }
  
    @Test
    public void testSubClassStoresSuperInterfaceValues() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SubClass extended = handler.as(SubClass.class);
  
      extended.setString("parentValue");
      assertEquals("parentValue", extended.getString());
    }
  
    @Test
    public void testUpCastRetainsSuperInterfaceValues() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SubClass extended = handler.as(SubClass.class);
  
      extended.setString("parentValue");
      Simple simple = extended.as(Simple.class);
      assertEquals("parentValue", simple.getString());
    }
  
    @Test
    public void testUpCastRetainsSubClassValues() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SubClass extended = handler.as(SubClass.class);
  
      extended.setExtended("subClassValue");
      SubClass extended2 = extended.as(Simple.class).as(SubClass.class);
      assertEquals("subClassValue", extended2.getExtended());
    }
  
    /** A test interface that is a sibling to {@link SubClass}. */
    public interface Sibling extends Simple {
      String getSibling();
  
      void setSibling(String value);
    }
  
    @Test
    public void testAsSiblingRetainsSuperInterfaceValues() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SubClass extended = handler.as(SubClass.class);
  
      extended.setString("parentValue");
      Sibling sibling = extended.as(Sibling.class);
      assertEquals("parentValue", sibling.getString());
    }
  
    /** A test interface that has the same methods as the parent. */
    public interface MethodConflict extends Simple {
      @Override
      String getString();
  
      @Override
      void setString(String value);
    }
  
    @Test
    public void testMethodConflictProvidesSameValue() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      MethodConflict methodConflict = handler.as(MethodConflict.class);
  
      methodConflict.setString("conflictValue");
      assertEquals("conflictValue", methodConflict.getString());
      assertEquals("conflictValue", methodConflict.as(Simple.class).getString());
    }
  
    /** A test interface that has the same methods as its parent and grandparent. */
    public interface DeepMethodConflict extends MethodConflict {
      @Override
      String getString();
  
      @Override
      void setString(String value);
  
      @Override
      int getPrimitive();
  
      @Override
      void setPrimitive(int value);
    }
  
    @Test
    public void testDeepMethodConflictProvidesSameValue() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      DeepMethodConflict deepMethodConflict = handler.as(DeepMethodConflict.class);
  
      // Tests overriding an already overridden method
      deepMethodConflict.setString("conflictValue");
      assertEquals("conflictValue", deepMethodConflict.getString());
      assertEquals("conflictValue", deepMethodConflict.as(MethodConflict.class).getString());
      assertEquals("conflictValue", deepMethodConflict.as(Simple.class).getString());
  
      // Tests overriding a method from an ancestor class
      deepMethodConflict.setPrimitive(5);
      assertEquals(5, deepMethodConflict.getPrimitive());
      assertEquals(5, deepMethodConflict.as(MethodConflict.class).getPrimitive());
      assertEquals(5, deepMethodConflict.as(Simple.class).getPrimitive());
    }
  
    /** A test interface that shares the same methods as {@link Sibling}. */
    public interface SimpleSibling extends PipelineOptions {
      String getString();
  
      void setString(String value);
    }
  
    @Test
    public void testDisjointSiblingsShareValues() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SimpleSibling proxy = handler.as(SimpleSibling.class);
      proxy.setString("siblingValue");
      assertEquals("siblingValue", proxy.getString());
      assertEquals("siblingValue", proxy.as(Simple.class).getString());
    }
  
    /** A test interface that joins two sibling interfaces that have conflicting methods. */
    public interface SiblingMethodConflict extends Simple, SimpleSibling {}
  
    @Test
    public void testSiblingMethodConflict() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      SiblingMethodConflict siblingMethodConflict = handler.as(SiblingMethodConflict.class);
      siblingMethodConflict.setString("siblingValue");
      assertEquals("siblingValue", siblingMethodConflict.getString());
      assertEquals("siblingValue", siblingMethodConflict.as(Simple.class).getString());
      assertEquals("siblingValue", siblingMethodConflict.as(SimpleSibling.class).getString());
    }
  
    /** A test interface that has only the getter and only a setter overriden. */
    public interface PartialMethodConflict extends Simple {
      @Override
      String getString();
  
      @Override
      void setPrimitive(int value);
    }
  
    @Test
    public void testPartialMethodConflictProvidesSameValue() throws Exception {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      PartialMethodConflict partialMethodConflict = handler.as(PartialMethodConflict.class);
  
      // Tests overriding a getter property that is only partially bound
      partialMethodConflict.setString("conflictValue");
      assertEquals("conflictValue", partialMethodConflict.getString());
      assertEquals("conflictValue", partialMethodConflict.as(Simple.class).getString());
  
      // Tests overriding a setter property that is only partially bound
      partialMethodConflict.setPrimitive(5);
      assertEquals(5, partialMethodConflict.getPrimitive());
      assertEquals(5, partialMethodConflict.as(Simple.class).getPrimitive());
    }
  
    @Test
    public void testResetRegistry() {
      Set<Class<? extends PipelineOptions>> defaultRegistry =
          new HashSet<>(PipelineOptionsFactory.getRegisteredOptions());
      assertThat(defaultRegistry, not(hasItem(FooOptions.class)));
  
      PipelineOptionsFactory.register(FooOptions.class);
      assertThat(PipelineOptionsFactory.getRegisteredOptions(), hasItem(FooOptions.class));
  
      PipelineOptionsFactory.resetCache();
      assertEquals(defaultRegistry, PipelineOptionsFactory.getRegisteredOptions());
    }
  
    @Test
    public void testJsonConversionForDefault() throws Exception {
      PipelineOptions options = PipelineOptionsFactory.create();
      assertNotNull(serializeDeserialize(PipelineOptions.class, options));
    }
  
    /** Test interface for JSON conversion of simple types. */
    public interface SimpleTypes extends PipelineOptions {
      int getInteger();
  
      void setInteger(int value);
  
      String getString();
  
      void setString(String value);
    }
  
    @Test
    public void testJsonConversionForSimpleTypes() throws Exception {
      SimpleTypes options = PipelineOptionsFactory.as(SimpleTypes.class);
      options.setString("TestValue");
      options.setInteger(5);
      SimpleTypes options2 = serializeDeserialize(SimpleTypes.class, options);
      assertEquals(5, options2.getInteger());
      assertEquals("TestValue", options2.getString());
    }
  
    @Test
    public void testJsonConversionOfAJsonConvertedType() throws Exception {
      SimpleTypes options = PipelineOptionsFactory.as(SimpleTypes.class);
      options.setString("TestValue");
      options.setInteger(5);
      // It is important here that our first serialization goes to our most basic
      // type so that we handle the case when we don't know the types of certain
      // properties because the intermediate instance of PipelineOptions never
      // saw their interface.
      SimpleTypes options2 =
          serializeDeserialize(
              SimpleTypes.class, serializeDeserialize(PipelineOptions.class, options));
      assertEquals(5, options2.getInteger());
      assertEquals("TestValue", options2.getString());
    }
  
    @Test
    public void testJsonConversionForPartiallySerializedValues() throws Exception {
      SimpleTypes options = PipelineOptionsFactory.as(SimpleTypes.class);
      options.setInteger(5);
      SimpleTypes options2 = serializeDeserialize(SimpleTypes.class, options);
      options2.setString("TestValue");
      SimpleTypes options3 = serializeDeserialize(SimpleTypes.class, options2);
      assertEquals(5, options3.getInteger());
      assertEquals("TestValue", options3.getString());
    }
  
    @Test
    public void testJsonConversionForOverriddenSerializedValues() throws Exception {
      SimpleTypes options = PipelineOptionsFactory.as(SimpleTypes.class);
      options.setInteger(-5);
      options.setString("NeedsToBeOverridden");
      SimpleTypes options2 = serializeDeserialize(SimpleTypes.class, options);
      options2.setInteger(5);
      options2.setString("TestValue");
      SimpleTypes options3 = serializeDeserialize(SimpleTypes.class, options2);
      assertEquals(5, options3.getInteger());
      assertEquals("TestValue", options3.getString());
    }
  
    /** Test interface for JSON conversion of container types. */
    public interface ContainerTypes extends PipelineOptions {
      List<String> getList();
  
      void setList(List<String> values);
  
      Map<String, String> getMap();
  
      void setMap(Map<String, String> values);
  
      Set<String> getSet();
  
      void setSet(Set<String> values);
    }
  
    @Test
    public void testJsonConversionForContainerTypes() throws Exception {
      List<String> list = ImmutableList.of("a", "b", "c");
      Map<String, String> map = ImmutableMap.of("d", "x", "e", "y", "f", "z");
      Set<String> set = ImmutableSet.of("g", "h", "i");
      ContainerTypes options = PipelineOptionsFactory.as(ContainerTypes.class);
      options.setList(list);
      options.setMap(map);
      options.setSet(set);
      ContainerTypes options2 = serializeDeserialize(ContainerTypes.class, options);
      assertEquals(list, options2.getList());
      assertEquals(map, options2.getMap());
      assertEquals(set, options2.getSet());
    }
  
    /** Test interface for conversion of inner types. */
    public static class InnerType {
      public double doubleField;
  
      static InnerType of(double value) {
        InnerType rval = new InnerType();
        rval.doubleField = value;
        return rval;
      }
  
      @Override
      public int hashCode() {
        return 0;
      }
  
      @Override
      public boolean equals(@Nullable Object obj) {
        return obj != null
            && getClass().equals(obj.getClass())
            && doubleField == ((InnerType) obj).doubleField;
      }
    }
  
    /** Test interface for conversion of generics and inner types. */
    public static class ComplexType {
      public String stringField;
      public Integer intField;
      public List<InnerType> genericType;
      public InnerType innerType;
  
      @Override
      public int hashCode() {
        return 0;
      }
  
      @Override
      public boolean equals(@Nullable Object obj) {
        return obj != null
            && getClass().equals(obj.getClass())
            && Objects.equals(stringField, ((ComplexType) obj).stringField)
            && Objects.equals(intField, ((ComplexType) obj).intField)
            && Objects.equals(genericType, ((ComplexType) obj).genericType)
            && Objects.equals(innerType, ((ComplexType) obj).innerType);
      }
    }
  
    /** Test interface. */
    public interface ComplexTypes extends PipelineOptions {
      ComplexType getComplexType();
  
      void setComplexType(ComplexType value);
    }
  
    @Test
    public void testJsonConversionForComplexType() throws Exception {
      ComplexType complexType = new ComplexType();
      complexType.stringField = "stringField";
      complexType.intField = 12;
      complexType.innerType = InnerType.of(12);
      complexType.genericType = ImmutableList.of(InnerType.of(16234), InnerType.of(24));
  
      ComplexTypes options = PipelineOptionsFactory.as(ComplexTypes.class);
      options.setComplexType(complexType);
      ComplexTypes options2 = serializeDeserialize(ComplexTypes.class, options);
      assertEquals(complexType, options2.getComplexType());
    }
  
    /** Test interface for testing ignored properties during serialization. */
    public interface IgnoredProperty extends PipelineOptions {
      @JsonIgnore
      String getValue();
  
      void setValue(String value);
    }
  
    @Test
    public void testJsonConversionOfIgnoredProperty() throws Exception {
      IgnoredProperty options = PipelineOptionsFactory.as(IgnoredProperty.class);
      options.setValue("TestValue");
  
      IgnoredProperty options2 = serializeDeserialize(IgnoredProperty.class, options);
      assertNull(options2.getValue());
    }
  
    /** Test class that is not serializable by Jackson. */
    public static class NotSerializable {
      private String value;
  
      public NotSerializable(String value) {
        this.value = value;
      }
  
      public String getValue() {
        return value;
      }
    }
  
    /** Test interface containing a class that is not serializable by Jackson. */
    public interface NotSerializableProperty extends PipelineOptions {
      NotSerializable getValue();
  
      void setValue(NotSerializable value);
    }
  
    @Test
    public void testJsonConversionOfNotSerializableProperty() throws Exception {
      NotSerializableProperty options = PipelineOptionsFactory.as(NotSerializableProperty.class);
      options.setValue(new NotSerializable("TestString"));
  
      expectedException.expect(JsonMappingException.class);
      expectedException.expectMessage("Failed to serialize and deserialize property 'value'");
      serializeDeserialize(NotSerializableProperty.class, options);
    }
  
    /**
     * Test interface that has {@link JsonIgnore @JsonIgnore} on a property that Jackson can't
     * serialize.
     */
    public interface IgnoredNotSerializableProperty extends PipelineOptions {
      @JsonIgnore
      NotSerializable getValue();
  
      void setValue(NotSerializable value);
    }
  
    @Test
    public void testJsonConversionOfIgnoredNotSerializableProperty() throws Exception {
      IgnoredNotSerializableProperty options =
          PipelineOptionsFactory.as(IgnoredNotSerializableProperty.class);
      options.setValue(new NotSerializable("TestString"));
  
      IgnoredNotSerializableProperty options2 =
          serializeDeserialize(IgnoredNotSerializableProperty.class, options);
      assertNull(options2.getValue());
    }
  
    /** Test class that is only serializable by Jackson with the added metadata. */
    public static class SerializableWithMetadata {
      private String value;
  
      public SerializableWithMetadata(@JsonProperty("value") String value) {
        this.value = value;
      }
  
      @JsonProperty("value")
      public String getValue() {
        return value;
      }
    }
  
    /**
     * Test interface containing a property that is serializable by Jackson only with the additional
     * metadata.
     */
    public interface SerializableWithMetadataProperty extends PipelineOptions {
      SerializableWithMetadata getValue();
  
      void setValue(SerializableWithMetadata value);
    }
  
    @Test
    public void testJsonConversionOfSerializableWithMetadataProperty() throws Exception {
      SerializableWithMetadataProperty options =
          PipelineOptionsFactory.as(SerializableWithMetadataProperty.class);
      options.setValue(new SerializableWithMetadata("TestString"));
  
      SerializableWithMetadataProperty options2 =
          serializeDeserialize(SerializableWithMetadataProperty.class, options);
      assertEquals("TestString", options2.getValue().getValue());
    }
  
    @Test
    public void testDisplayDataItemProperties() {
      PipelineOptions options = PipelineOptionsFactory.create();
      options.setTempLocation("myTemp");
      DisplayData displayData = DisplayData.from(options);
  
      assertThat(
          displayData,
          hasDisplayItem(
              allOf(
                  hasKey("tempLocation"),
                  hasType(DisplayData.Type.STRING),
                  hasValue("myTemp"),
                  hasNamespace(PipelineOptions.class))));
    }
  
    @Test
    public void testDisplayDataTypes() {
      Instant now = Instant.now();
  
      TypedOptions options = PipelineOptionsFactory.as(TypedOptions.class);
      options.setInteger(1234);
      options.setTimestamp(now);
      options.setJavaClass(ProxyInvocationHandlerTest.class);
      options.setObject(
          new Serializable() {
            @Override
            public String toString() {
              return "foobar";
            }
          });
  
      DisplayData displayData = DisplayData.from(options);
  
      assertThat(displayData, hasDisplayItem("integer", 1234));
      assertThat(displayData, hasDisplayItem("timestamp", now));
      assertThat(displayData, hasDisplayItem("javaClass", ProxyInvocationHandlerTest.class));
      assertThat(displayData, hasDisplayItem("object", "foobar"));
    }
  
    /** Test interface. */
    public interface TypedOptions extends PipelineOptions {
      int getInteger();
  
      void setInteger(int value);
  
      Instant getTimestamp();
  
      void setTimestamp(Instant value);
  
      Class<?> getJavaClass();
  
      void setJavaClass(Class<?> value);
  
      Object getObject();
  
      void setObject(Object value);
    }
  
    @Test
    @Category(NeedsRunner.class)
    public void pipelineOptionsDisplayDataExceptionShouldFail() {
      Object brokenValueType =
          new Object() {
            @JsonValue
            public int getValue() {
              return 42;
            }
  
            @Override
            public String toString() {
              throw new RuntimeException("oh noes!!");
            }
          };
  
      p.getOptions().as(ObjectPipelineOptions.class).setValue(brokenValueType);
  
      p.apply(Create.of(1, 2, 3));
  
      expectedException.expectMessage(
          ProxyInvocationHandler.PipelineOptionsDisplayData.class.getName());
      expectedException.expectMessage("oh noes!!");
      p.run();
    }
  
    /** {@link PipelineOptions} to inject bad object implementations. */
    public interface ObjectPipelineOptions extends PipelineOptions {
      Object getValue();
  
      void setValue(Object value);
    }
  
    public interface PrimitiveIntOptions extends PipelineOptions {
      int getInt();
  
      void setInt(int value);
    }
  
    @Test
    public void testPrimitiveIntegerFromJsonOptions() throws Exception {
      String optionsJson =
          "{\"options\":{\"appName\":\"ProxyInvocationHandlerTest\",\"optionsId\":1,\"int\":\"100\"},\"display_data\":[{\"namespace\":\"org.apache.beam.sdk.options.ProxyInvocationHandlerTest$DisplayDataOptions\",\"key\":\"int\",\"type\":\"INTEGER\",\"value\":100},{\"namespace\":\"org.apache.beam.sdk.options.ApplicationNameOptions\",\"key\":\"appName\",\"type\":\"STRING\",\"value\":\"ProxyInvocationHandlerTest\"}]}";
  
      PrimitiveIntOptions options =
          MAPPER.readValue(optionsJson, PipelineOptions.class).as(PrimitiveIntOptions.class);
  
      int value = options.getInt();
      assertEquals(100, value);
    }
  
    @Test
    public void testDisplayDataInheritanceNamespace() {
      ExtendsBaseOptions options = PipelineOptionsFactory.as(ExtendsBaseOptions.class);
      options.setFoo("bar");
  
      DisplayData displayData = DisplayData.from(options);
  
      assertThat(
          displayData,
          hasDisplayItem(
              allOf(hasKey("foo"), hasValue("bar"), hasNamespace(ExtendsBaseOptions.class))));
    }
  
    /** Test interface. */
    public interface BaseOptions extends PipelineOptions {
      String getFoo();
  
      void setFoo(String value);
    }
  
    /** Test interface. */
    public interface ExtendsBaseOptions extends BaseOptions {
      @Override
      String getFoo();
  
      @Override
      void setFoo(String value);
    }
  
    @Test
    public void testDisplayDataExcludedFromOverriddenBaseClass() {
      ExtendsBaseOptions options = PipelineOptionsFactory.as(ExtendsBaseOptions.class);
      options.setFoo("bar");
  
      DisplayData displayData = DisplayData.from(options);
      assertThat(displayData, not(hasDisplayItem(hasNamespace(BaseOptions.class))));
    }
  
    @Test
    public void testDisplayDataIncludedForDisjointInterfaceHierarchies() {
      FooOptions fooOptions = PipelineOptionsFactory.as(FooOptions.class);
      fooOptions.setFoo("foo");
  
      BarOptions barOptions = fooOptions.as(BarOptions.class);
      barOptions.setBar("bar");
  
      DisplayData data = DisplayData.from(barOptions);
      assertThat(data, hasDisplayItem(allOf(hasKey("foo"), hasNamespace(FooOptions.class))));
      assertThat(data, hasDisplayItem(allOf(hasKey("bar"), hasNamespace(BarOptions.class))));
    }
  
    /** Test interface. */
    public interface FooOptions extends PipelineOptions {
      String getFoo();
  
      void setFoo(String value);
    }
  
    /** Test interface. */
    public interface BarOptions extends PipelineOptions {
      String getBar();
  
      void setBar(String value);
    }
  
    public static class JacksonObject {
      String value;
    }
  
    public static class JacksonObjectSerializer extends StdSerializer<JacksonObject> {
      public JacksonObjectSerializer() {
        super(JacksonObject.class);
      }
  
      @Override
      public void serialize(JacksonObject value, JsonGenerator gen, SerializerProvider provider)
          throws IOException {
        gen.writeString(value.value);
      }
    }
  
    public static class JacksonObjectDeserializer extends StdDeserializer<JacksonObject> {
      public JacksonObjectDeserializer() {
        super(JacksonObject.class);
      }
  
      @Override
      public JacksonObject deserialize(JsonParser p, DeserializationContext ctxt)
          throws IOException, JsonProcessingException {
        JacksonObject obj = new JacksonObject();
        obj.value = p.getValueAsString();
        return obj;
      }
    }
  
    public interface JacksonOptions extends PipelineOptions {
      @JsonSerialize(using = JacksonObjectSerializer.class)
      @JsonDeserialize(using = JacksonObjectDeserializer.class)
      JacksonObject getJacksonObject();
  
      void setJacksonObject(JacksonObject value);
    }
  
    @Test
    public void testJacksonSerializeAndDeserialize() throws Exception {
      JacksonOptions options = PipelineOptionsFactory.as(JacksonOptions.class);
      JacksonObject value = new JacksonObject();
      value.value = "foo";
  
      options.setJacksonObject(value);
  
      JacksonOptions deserializedOptions = serializeDeserialize(JacksonOptions.class, options);
      assertEquals(options.getJacksonObject().value, deserializedOptions.getJacksonObject().value);
    }
  
    @Test
    public void testDisplayDataExcludesDefaultValues() {
      PipelineOptions options = PipelineOptionsFactory.as(HasDefaults.class);
      DisplayData data = DisplayData.from(options);
  
      assertThat(data, not(hasDisplayItem("foo")));
    }
  
    /** Test interface. */
    public interface HasDefaults extends PipelineOptions {
      @Default.String("bar")
      String getFoo();
  
      void setFoo(String value);
    }
  
    @Test
    public void testDisplayDataExcludesValuesAccessedButNeverSet() {
      HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
      assertEquals("bar", options.getFoo());
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, not(hasDisplayItem("foo")));
    }
  
    @Test
    public void testDisplayDataExcludesHiddenValues() {
      HasHidden options = PipelineOptionsFactory.as(HasHidden.class);
      options.setFoo("bar");
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, not(hasDisplayItem("foo")));
    }
  
    /** Test interface. */
    public interface HasHidden extends PipelineOptions {
      @Hidden
      String getFoo();
  
      void setFoo(String value);
    }
  
    @Test
    public void testDisplayDataIncludesExplicitlySetDefaults() {
      HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
      String defaultValue = options.getFoo();
      options.setFoo(defaultValue);
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, hasDisplayItem("foo", defaultValue));
    }
  
    @Test
    public void testDisplayDataNullValuesConvertedToEmptyString() throws Exception {
      FooOptions options = PipelineOptionsFactory.as(FooOptions.class);
      options.setFoo(null);
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, hasDisplayItem("foo", ""));
  
      FooOptions deserializedOptions = serializeDeserialize(FooOptions.class, options);
      DisplayData deserializedData = DisplayData.from(deserializedOptions);
      assertThat(deserializedData, hasDisplayItem("foo", ""));
    }
  
    @Test
    public void testDisplayDataArrayValue() throws Exception {
      ArrayOptions options = PipelineOptionsFactory.as(ArrayOptions.class);
      options.setDeepArray(new String[][] {new String[] {"a", "b"}, new String[] {"c"}});
      options.setDeepPrimitiveArray(new int[][] {new int[] {1, 2}, new int[] {3}});
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, hasDisplayItem("deepArray", "[[a, b], [c]]"));
      assertThat(data, hasDisplayItem("deepPrimitiveArray", "[[1, 2], [3]]"));
  
      ArrayOptions deserializedOptions = serializeDeserialize(ArrayOptions.class, options);
      DisplayData deserializedData = DisplayData.from(deserializedOptions);
      assertThat(deserializedData, hasDisplayItem("deepPrimitiveArray", "[[1, 2], [3]]"));
    }
  
    /** Test interface. */
    public interface ArrayOptions extends PipelineOptions {
      String[][] getDeepArray();
  
      void setDeepArray(String[][] value);
  
      int[][] getDeepPrimitiveArray();
  
      void setDeepPrimitiveArray(int[][] value);
    }
  
    @Test
    public void testDisplayDataJsonSerialization() throws IOException {
      FooOptions options = PipelineOptionsFactory.as(FooOptions.class);
      options.setFoo("bar");
  
      @SuppressWarnings("unchecked")
      Map<String, Object> map = MAPPER.readValue(MAPPER.writeValueAsBytes(options), Map.class);
  
      assertThat("main pipeline options data keyed as 'options'", map, Matchers.hasKey("options"));
      assertThat("display data keyed as 'display_data'", map, Matchers.hasKey("display_data"));
  
      Map<?, ?> expectedDisplayItem =
          ImmutableMap.<String, String>builder()
              .put("namespace", FooOptions.class.getName())
              .put("key", "foo")
              .put("value", "bar")
              .put("type", "STRING")
              .build();
  
      @SuppressWarnings("unchecked")
      List<Map<?, ?>> deserializedDisplayData = (List<Map<?, ?>>) map.get("display_data");
      assertThat(deserializedDisplayData, hasItem(expectedDisplayItem));
    }
  
    @Test
    public void testDisplayDataFromDeserializedJson() throws Exception {
      FooOptions options = PipelineOptionsFactory.as(FooOptions.class);
      options.setFoo("bar");
      DisplayData data = DisplayData.from(options);
      assertThat(data, hasDisplayItem("foo", "bar"));
  
      FooOptions deserializedOptions = serializeDeserialize(FooOptions.class, options);
      DisplayData dataAfterDeserialization = DisplayData.from(deserializedOptions);
      assertEquals(data, dataAfterDeserialization);
    }
  
    @Test
    public void testDisplayDataDeserializationWithRegistration() throws Exception {
      PipelineOptionsFactory.register(HasClassOptions.class);
      HasClassOptions options = PipelineOptionsFactory.as(HasClassOptions.class);
      options.setClassOption(ProxyInvocationHandlerTest.class);
  
      PipelineOptions deserializedOptions = serializeDeserialize(PipelineOptions.class, options);
      DisplayData displayData = DisplayData.from(deserializedOptions);
      assertThat(displayData, hasDisplayItem("classOption", ProxyInvocationHandlerTest.class));
    }
  
    @Test
    public void testDisplayDataMissingPipelineOptionsRegistration() throws Exception {
      HasClassOptions options = PipelineOptionsFactory.as(HasClassOptions.class);
      options.setClassOption(ProxyInvocationHandlerTest.class);
  
      PipelineOptions deserializedOptions = serializeDeserialize(PipelineOptions.class, options);
      DisplayData displayData = DisplayData.from(deserializedOptions);
      String expectedJsonValue = MAPPER.writeValueAsString(ProxyInvocationHandlerTest.class);
      assertThat(displayData, hasDisplayItem("classOption", expectedJsonValue));
    }
  
    /** Test interface. */
    public interface HasClassOptions extends PipelineOptions {
      Class<?> getClassOption();
  
      void setClassOption(Class<?> value);
    }
  
    @Test
    public void testDisplayDataJsonValueSetAfterDeserialization() throws Exception {
      FooOptions options = PipelineOptionsFactory.as(FooOptions.class);
      options.setFoo("bar");
      DisplayData data = DisplayData.from(options);
      assertThat(data, hasDisplayItem("foo", "bar"));
  
      FooOptions deserializedOptions = serializeDeserialize(FooOptions.class, options);
      deserializedOptions.setFoo("baz");
      DisplayData dataAfterDeserialization = DisplayData.from(deserializedOptions);
      assertThat(dataAfterDeserialization, hasDisplayItem("foo", "baz"));
    }
  
    private <T extends PipelineOptions> T serializeDeserialize(Class<T> kls, PipelineOptions options)
        throws Exception {
      String value = MAPPER.writeValueAsString(options);
      return MAPPER.readValue(value, PipelineOptions.class).as(kls);
    }
  
    @Test
    public void testDisplayDataExcludesJsonIgnoreOptions() {
      IgnoredProperty options = PipelineOptionsFactory.as(IgnoredProperty.class);
      options.setValue("foobar");
  
      DisplayData data = DisplayData.from(options);
      assertThat(data, not(hasDisplayItem("value")));
    }
  
    private static class CapturesOptions implements Serializable {
      PipelineOptions options = PipelineOptionsFactory.create();
    }
  
    @Test
    public void testOptionsAreNotSerializable() {
      expectedException.expectCause(Matchers.instanceOf(NotSerializableException.class));
      SerializableUtils.clone(new CapturesOptions());
    }
  
    @Test
    public void testGetOptionNameFromMethod() throws NoSuchMethodException {
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      handler.as(BaseOptions.class);
      assertEquals("foo", handler.getOptionName(BaseOptions.class.getMethod("getFoo")));
    }
  
    private DynamicType.Unloaded<? extends PipelineOptions> spinNewInterface(int methodNumber) {
      return new ByteBuddy()
          .makeInterface(PipelineOptions.class)
          .defineMethod("getDynamicMethod" + methodNumber, String.class, Visibility.PUBLIC)
          .withoutCode()
          .defineMethod("setDynamicMethod" + methodNumber, Void.TYPE, Visibility.PUBLIC)
          .withParameters(String.class)
          .withoutCode()
          .make();
    }
  
    private Map<Integer, Class<? extends PipelineOptions>> loadAllInterfaces(
        int numInterfaces, ClassLoader classLoader) {
      List<DynamicType.Unloaded<? extends PipelineOptions>> dynamicInterfaces =
          IntStream.range(0, numInterfaces)
              .mapToObj(this::spinNewInterface)
              .collect(Collectors.toList());
  
      DynamicType.Loaded<Object> root =
          new ByteBuddy().subclass(Object.class).make().include(dynamicInterfaces).load(classLoader);
  
      Map<TypeDescription, Class<?>> loadedInterfaces = root.getLoadedAuxiliaryTypes();
      Map<Integer, Class<? extends PipelineOptions>> result = Maps.newHashMap();
  
      IntStream.range(0, numInterfaces)
          .forEach(
              i -> {
                DynamicType.Unloaded<? extends PipelineOptions> iface = dynamicInterfaces.get(i);
                Class<?> clazz = loadedInterfaces.get(iface.getTypeDescription());
                result.put(i, (Class<? extends PipelineOptions>) clazz);
              });
  
      return result;
    }
  
    @Test
    public void testConcurrency() throws Exception {
      int numInterfaces = 100;
      int numWorkers = 10;
      int numReaders = 10;
      int step = numInterfaces / numWorkers;
  
      ProxyInvocationHandler handler = new ProxyInvocationHandler(Maps.newHashMap());
      ClassLoader cl = Thread.currentThread().getContextClassLoader();
      Map<Integer, Class<? extends PipelineOptions>> ifaces = loadAllInterfaces(numInterfaces, cl);
      CountDownLatch startWaiter = new CountDownLatch(1);
      CountDownLatch done = new CountDownLatch(numWorkers);
  
      ExecutorService executor = Executors.newFixedThreadPool(numWorkers + numReaders);
      List<Future<?>> futs = Lists.newArrayList();
  
      // launch `numWorkers` concurrent "writers" that will try to cast the proxy handler to various
      // interfaces and set a value on them.
      for (int start = 0; start < numInterfaces; start += step) {
        final int s = start;
        final int end = start + step;
        Callable<Void> worker =
            () -> {
              startWaiter.await();
              for (int i = s; i < end; i++) {
                Class<? extends PipelineOptions> iface = ifaces.get(i);
                Method setter = iface.getDeclaredMethod("setDynamicMethod" + i, String.class);
                PipelineOptions opt1 = handler.as(iface);
                setter.invoke(opt1, "test-" + i);
              }
              done.countDown();
              return null;
            };
        futs.add(executor.submit(worker));
      }
  
      // launch concurrent readers that call `toString` and serializes the options, making sure they
      // don't fail.
      for (int i = 0; i < 10; i++) {
        Callable<Void> worker =
            () -> {
              while (true) {
                if (done.getCount() == 0) {
                  return null;
                }
                assertNotNull(handler.toString());
                PipelineOptionsFactory.MAPPER.writeValue(
                    ByteStreams.nullOutputStream(), handler.as(PipelineOptions.class));
              }
            };
        futs.add(executor.submit(worker));
      }
  
      // wait for everything to finish
      startWaiter.countDown();
      for (Future<?> fut : futs) {
        fut.get();
      }
  
      // ensure that the serialized json contains every value we set.
      TokenBuffer tokenBuffer = new TokenBuffer(PipelineOptionsFactory.MAPPER, false);
      PipelineOptionsFactory.MAPPER.writeValue(tokenBuffer, handler.as(PipelineOptions.class));
  
      JsonNode tree = PipelineOptionsFactory.MAPPER.readTree(tokenBuffer.asParser());
      JsonNode optionsNode = tree.get("options");
      for (int i = 0; i < numInterfaces; i++) {
        JsonNode methodNode = optionsNode.get("dynamicMethod" + i);
        assertNotNull(methodNode);
        assertEquals("test-" + i, methodNode.asText());
      }
    }
  }
  