static Builder builder() {
    return new AutoValue_PubsubRowToMessage.Builder();
  }
@Test
  public void testExpandOutput() {
    PayloadSerializer jsonPayloadSerializer =
        new JsonPayloadSerializerProvider().getSerializer(ALL_DATA_TYPES_SCHEMA, ImmutableMap.of());

    Instant embeddedTimestamp = Instant.now();
    Instant mockTimestamp = Instant.ofEpochMilli(embeddedTimestamp.getMillis() + 1000000L);
    String embeddedTimestampString = embeddedTimestamp.toString();
    String mockTimestampString = mockTimestamp.toString();

    byte[] bytes =
        "All the 🌎's a 🎦, and all the 🚹 and 🚺 merely players. They 🈷️their 🚪and their 🎟️and 1️⃣🚹in his time ▶️many🤺🪂🏆"
            .getBytes(StandardCharsets.UTF_8);

    Field attributesField = Field.of(DEFAULT_ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE);
    Field timestampField = Field.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE);
    Field payloadBytesField = Field.of(DEFAULT_PAYLOAD_KEY_NAME, FieldType.BYTES);
    Field payloadRowField =
        Field.of(DEFAULT_PAYLOAD_KEY_NAME, FieldType.row(ALL_DATA_TYPES_SCHEMA));

    Row withAllDataTypes =
        rowWithAllDataTypes(
            true,
            (byte) 0,
            Instant.now().toDateTime(),
            BigDecimal.valueOf(1L),
            3.12345,
            4.1f,
            (short) 5,
            2,
            7L,
            "asdfjkl;");

    byte[] serializedWithAllDataTypes = jsonPayloadSerializer.serialize(withAllDataTypes);

    Schema withAttributesSchema = Schema.of(attributesField);
    Row withAttributes =
        Row.withSchema(withAttributesSchema).attachValues(ImmutableMap.of("aaa", "111"));

    Schema withTimestampSchema = Schema.of(timestampField);
    Row withEmbeddedTimestamp = Row.withSchema(withTimestampSchema).attachValues(embeddedTimestamp);

    Schema withPayloadBytesSchema = Schema.of(payloadBytesField);
    Row withPayloadBytes = Row.withSchema(withPayloadBytesSchema).attachValues(bytes);

    Schema withPayloadRowSchema = Schema.of(payloadRowField);
    Row withPayloadRow = Row.withSchema(withPayloadRowSchema).attachValues(withAllDataTypes);

    PubsubRowToMessage transformWithoutSerializer =
        PubsubRowToMessage.builder().setMockInstant(mockTimestamp).build();

    PubsubRowToMessage transformWithSerializer =
        PubsubRowToMessage.builder()
            .setMockInstant(mockTimestamp)
            .setPayloadSerializer(jsonPayloadSerializer)
            .build();

    PCollection<Row> a0t0p0u1 =
        pipeline.apply(Create.of(withAllDataTypes)).setRowSchema(ALL_DATA_TYPES_SCHEMA);
    PubsubMessage a0t0p0u1Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, mockTimestampString));
    PAssert.that(
            "expected: payload: user serialized row, attributes: ParDo mocked generated timestamp only",
            a0t0p0u1.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a0t0p0u1Message);

    PCollection<Row> a1t0p0u1 =
        pipeline
            .apply(Create.of(merge(withAttributes, withAllDataTypes)))
            .setRowSchema(merge(withAttributesSchema, ALL_DATA_TYPES_SCHEMA));
    PubsubMessage a1t0p0u1Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of("aaa", "111", DEFAULT_EVENT_TIMESTAMP_KEY_NAME, mockTimestampString));
    PAssert.that(
            "expected: payload: user serialized row, attributes: input attributes and ParDo generated timestamp",
            a1t0p0u1.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a1t0p0u1Message);

    PCollection<Row> a0t1p0u1 =
        pipeline
            .apply(Create.of(merge(withEmbeddedTimestamp, withAllDataTypes)))
            .setRowSchema(merge(withTimestampSchema, ALL_DATA_TYPES_SCHEMA));
    PubsubMessage a0t1p0u1Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, embeddedTimestampString));
    PAssert.that(
            "expected: payload: user serialized row, attributes: timestamp from row input",
            a0t1p0u1.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a0t1p0u1Message);

    PCollection<Row> a0t0p1bu0 =
        pipeline.apply(Create.of(withPayloadBytes)).setRowSchema(withPayloadBytesSchema);
    PubsubMessage a0t0p1bu0Message =
        new PubsubMessage(
            bytes, ImmutableMap.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, mockTimestampString));
    PAssert.that(
            "expected: payload: raw bytes, attributes: ParDo generated timestamp",
            a0t0p1bu0.apply(transformWithoutSerializer).get(OUTPUT))
        .containsInAnyOrder(a0t0p1bu0Message);

    PCollection<Row> a0t0p1ru0 =
        pipeline.apply(Create.of(withPayloadRow)).setRowSchema(withPayloadRowSchema);
    PubsubMessage a0t0p1ru0Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, mockTimestampString));
    PAssert.that(
            "expected: payload: non-user row, attributes: ParDo generated timestamp",
            a0t0p1ru0.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a0t0p1ru0Message);

    PCollection<Row> a1t0p1bu0 =
        pipeline
            .apply(Create.of(merge(withAttributes, withPayloadBytes)))
            .setRowSchema(merge(withAttributesSchema, withPayloadBytesSchema));
    PubsubMessage a1t0p1bu0Message =
        new PubsubMessage(
            bytes,
            ImmutableMap.of("aaa", "111", DEFAULT_EVENT_TIMESTAMP_KEY_NAME, mockTimestampString));
    PAssert.that(
            "expected: raw bytes, attributes: embedded attributes and ParDo generated timestamp",
            a1t0p1bu0.apply(transformWithoutSerializer).get(OUTPUT))
        .containsInAnyOrder(a1t0p1bu0Message);

    PCollection<Row> a0t1p1bu0 =
        pipeline
            .apply(Create.of(merge(withEmbeddedTimestamp, withPayloadBytes)))
            .setRowSchema(merge(withTimestampSchema, withPayloadBytesSchema));
    PubsubMessage a0t1p1bu0Message =
        new PubsubMessage(
            bytes, ImmutableMap.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, embeddedTimestampString));
    PAssert.that(
            "expected: payload: raw bytes, attributes: timestamp from row input",
            a0t1p1bu0.apply(transformWithoutSerializer).get(OUTPUT))
        .containsInAnyOrder(a0t1p1bu0Message);

    PCollection<Row> a1t1p1bu0 =
        pipeline
            .apply(Create.of(merge(merge(withAttributes, withEmbeddedTimestamp), withPayloadBytes)))
            .setRowSchema(
                merge(merge(withAttributesSchema, withTimestampSchema), withPayloadBytesSchema));
    PubsubMessage a1t1p1bu0Message =
        new PubsubMessage(
            bytes,
            ImmutableMap.of(
                "aaa", "111", DEFAULT_EVENT_TIMESTAMP_KEY_NAME, embeddedTimestampString));
    PAssert.that(
            "expected: payload: raw bytes, attributes: from row input including its embedded timestamp",
            a1t1p1bu0.apply(transformWithoutSerializer).get(OUTPUT))
        .containsInAnyOrder(a1t1p1bu0Message);

    PCollection<Row> a1t1p1ru0 =
        pipeline
            .apply(Create.of(merge(merge(withAttributes, withEmbeddedTimestamp), withPayloadRow)))
            .setRowSchema(
                merge(merge(withAttributesSchema, withTimestampSchema), withPayloadRowSchema));
    PubsubMessage a1t1p1ru0Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of(
                "aaa", "111", DEFAULT_EVENT_TIMESTAMP_KEY_NAME, embeddedTimestampString));
    PAssert.that(
            "expected: payload: serialized row payload input, attributes: from row input including its embedded timestamp",
            a1t1p1ru0.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a1t1p1ru0Message);

    PCollection<Row> a1t1p0u1 =
        pipeline
            .apply(Create.of(merge(merge(withAttributes, withEmbeddedTimestamp), withAllDataTypes)))
            .setRowSchema(
                merge(merge(withAttributesSchema, withTimestampSchema), ALL_DATA_TYPES_SCHEMA));
    PubsubMessage a1t1p0u1Message =
        new PubsubMessage(
            serializedWithAllDataTypes,
            ImmutableMap.of(
                "aaa", "111", DEFAULT_EVENT_TIMESTAMP_KEY_NAME, embeddedTimestampString));
    PAssert.that(
            "expected: payload: serialized user fields, attributes: from row input including embedded timestamp",
            a1t1p0u1.apply(transformWithSerializer).get(OUTPUT))
        .containsInAnyOrder(a1t1p0u1Message);

    pipeline.run(PIPELINE_OPTIONS);
  }