SchemaCoder inconsistent equality behavior for POJO

import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.SchemaCoder;
import org.apache.beam.sdk.values.Row;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

import java.nio.charset.Charset;

import static org.junit.Assert.*;

@RunWith(JUnit4.class)
public class SchemaCoderTest {

  public static class Pojo {
    private final byte[] bytes;
    private final String id;

    public Pojo(byte[] bytes, String id) {
      this.bytes = bytes;
      this.id = id;
    }

    public byte[] getBytes() {
      return bytes;
    }

    public String getId() {
      return id;
    }
  }

  @Test
  public void testCoder() {
    Schema schema = Schema.builder().addByteArrayField("bytes").addStringField("id").build();

    SchemaCoder<Pojo> coder = SchemaCoder.<Pojo>of(
            schema,
            t -> Row.withSchema(schema).addValues(t.getBytes(), t.getId()).build(),
            r -> new Pojo(r.getBytes("bytes"), r.getString("id")));

    Pojo p1 = new Pojo("hello".getBytes(Charset.forName("UTF-8")), "world");
    Pojo p2 = new Pojo("hello".getBytes(Charset.forName("UTF-8")), "world");

    assertNotEquals(p1, p2); // EXPECTED, p1.equals(p2) == false

    assertFalse(coder.consistentWithEquals()); // FAIL, returns true
    assertEquals(coder.structuralValue(p1), coder.structuralValue(p2)); // FAIL
  }
}