@@ -0,0 +1,2085 @@
+package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;%0A%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.OutputStream;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0Aimport org.apache.beam.sdk.coders.Coder;%0Aimport org.apache.beam.sdk.coders.StructuredCoder;%0Aimport scala.Tuple2;%0A%0A/**%0A * Beam coder to encode/decode Tuple2 scala types.%0A * @param %3CT1%3E first field type parameter%0A * @param %3CT2%3E second field type parameter%0A */%0Apublic class Tuple2Coder%3CT1, T2%3E extends StructuredCoder%3CTuple2%3CT1, T2%3E%3E %7B%0A  private final Coder%3CT1%3E firstFieldCoder;%0A  private final Coder%3CT2%3E secondFieldCoder;%0A%0A  public static %3CK, V%3E Tuple2Coder%3CK, V%3E of(Coder%3CK%3E firstFieldCoder, Coder%3CV%3E secondFieldCoder) %7B%0A    return new Tuple2Coder%3C%3E(firstFieldCoder, secondFieldCoder);%0A  %7D%0A%0A  private Tuple2Coder(Coder%3CT1%3E firstFieldCoder, Coder%3CT2%3E secondFieldCoder) %7B%0A    this.firstFieldCoder = firstFieldCoder;%0A    this.secondFieldCoder = secondFieldCoder;%0A  %7D%0A%0A%0A  @Override public void encode(Tuple2%3CT1, T2%3E value, OutputStream outStream)%0A      throws IOException %7B%0A    firstFieldCoder.encode(value._1(), outStream);%0A    secondFieldCoder.encode(value._2(), outStream);%0A  %7D%0A%0A  @Override public Tuple2%3CT1, T2%3E decode(InputStream inStream) throws IOException %7B%0A    T1 firstField = firstFieldCoder.decode(inStream);%0A    T2 secondField = secondFieldCoder.decode(inStream);%0A    return Tuple2.apply(firstField, secondField);%0A  %7D%0A%0A  @Override public List%3C? extends Coder%3C?%3E%3E getCoderArguments() %7B%0A    return Arrays.asList(firstFieldCoder, secondFieldCoder);%0A  %7D%0A%0A  @Override%0A  public void verifyDeterministic() throws NonDeterministicException %7B%0A    verifyDeterministic(this, %22First field coder must be deterministic%22, firstFieldCoder);%0A    verifyDeterministic(this, %22Second field coder must be deterministic%22, secondFieldCoder);%0A  %7D%0A%0A  /** Returns the coder for first field. */%0A  public Coder%3CT1%3E getFirstFieldCoder() %7B%0A    return firstFieldCoder;%0A  %7D%0A%0A  /** Returns the coder for second field. */%0A  public Coder%3CT2%3E getSecondFieldCoder() %7B%0A    return secondFieldCoder;%0A  %7D%0A%7D%0A

Create a Tuple2Coder to encode scala tuple2