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.kafka.common.message;
  
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.errors.UnsupportedVersionException;
  import org.apache.kafka.common.protocol.ByteBufferAccessor;
  import org.apache.kafka.common.protocol.MessageUtil;
  import org.apache.kafka.common.protocol.ObjectSerializationCache;
  import org.apache.kafka.common.utils.ByteUtils;
  
  import com.fasterxml.jackson.databind.JsonNode;
  
  import org.junit.jupiter.api.Test;
  
  import java.nio.ByteBuffer;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.function.Consumer;
  
  import static org.apache.kafka.common.protocol.MessageUtil.UNSIGNED_INT_MAX;
  import static org.apache.kafka.common.protocol.MessageUtil.UNSIGNED_SHORT_MAX;
  import static org.junit.jupiter.api.Assertions.assertArrayEquals;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  
  public class SimpleExampleMessageTest {
  
      @Test
      public void shouldStoreField() {
          final Uuid uuid = Uuid.randomUuid();
          final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
  
          final SimpleExampleMessageData out = new SimpleExampleMessageData();
          out.setProcessId(uuid);
          out.setZeroCopyByteBuffer(buf);
  
          assertEquals(uuid, out.processId());
          assertEquals(buf, out.zeroCopyByteBuffer());
  
          out.setNullableZeroCopyByteBuffer(null);
          assertNull(out.nullableZeroCopyByteBuffer());
          out.setNullableZeroCopyByteBuffer(buf);
          assertEquals(buf, out.nullableZeroCopyByteBuffer());
      }
  
      @Test
      public void shouldThrowIfCannotWriteNonIgnorableField() {
          // processId is not supported in v0 and is not marked as ignorable
  
          final SimpleExampleMessageData out = new SimpleExampleMessageData().setProcessId(Uuid.randomUuid());
          assertThrows(UnsupportedVersionException.class, () ->
                  out.write(new ByteBufferAccessor(ByteBuffer.allocate(64)), new ObjectSerializationCache(), (short) 0));
      }
  
      @Test
      public void shouldDefaultField() {
          final SimpleExampleMessageData out = new SimpleExampleMessageData();
          assertEquals(Uuid.fromString("AAAAAAAAAAAAAAAAAAAAAA"), out.processId());
          assertEquals(ByteUtils.EMPTY_BUF, out.zeroCopyByteBuffer());
          assertEquals(ByteUtils.EMPTY_BUF, out.nullableZeroCopyByteBuffer());
      }
  
      @Test
      public void shouldRoundTripFieldThroughBuffer() {
          final Uuid uuid = Uuid.randomUuid();
          final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
          final SimpleExampleMessageData out = new SimpleExampleMessageData();
          out.setProcessId(uuid);
          out.setZeroCopyByteBuffer(buf);
  
          final ByteBuffer buffer = MessageUtil.toByteBufferAccessor(out, (short) 1).buffer();
  
          final SimpleExampleMessageData in = new SimpleExampleMessageData();
          in.read(new ByteBufferAccessor(buffer), (short) 1);
  
          buf.rewind();
  
          assertEquals(uuid, in.processId());
          assertEquals(buf, in.zeroCopyByteBuffer());
          assertEquals(ByteUtils.EMPTY_BUF, in.nullableZeroCopyByteBuffer());
      }
  
      @Test
      public void shouldRoundTripFieldThroughBufferWithNullable() {
          final Uuid uuid = Uuid.randomUuid();
          final ByteBuffer buf1 = ByteBuffer.wrap(new byte[] {1, 2, 3});
          final ByteBuffer buf2 = ByteBuffer.wrap(new byte[] {4, 5, 6});
          final SimpleExampleMessageData out = new SimpleExampleMessageData();
          out.setProcessId(uuid);
          out.setZeroCopyByteBuffer(buf1);
          out.setNullableZeroCopyByteBuffer(buf2);
  
          final ByteBuffer buffer = MessageUtil.toByteBufferAccessor(out, (short) 1).buffer();
  
          final SimpleExampleMessageData in = new SimpleExampleMessageData();
          in.read(new ByteBufferAccessor(buffer), (short) 1);
  
          buf1.rewind();
          buf2.rewind();
  
          assertEquals(uuid, in.processId());
          assertEquals(buf1, in.zeroCopyByteBuffer());
          assertEquals(buf2, in.nullableZeroCopyByteBuffer());
      }
  
      @Test
      public void shouldImplementEqualsAndHashCode() {
          final Uuid uuid = Uuid.randomUuid();
          final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
          final SimpleExampleMessageData a = new SimpleExampleMessageData();
          a.setProcessId(uuid);
          a.setZeroCopyByteBuffer(buf);
  
          final SimpleExampleMessageData b = new SimpleExampleMessageData();
          b.setProcessId(uuid);
          b.setZeroCopyByteBuffer(buf);
  
          assertEquals(a, b);
          assertEquals(a.hashCode(), b.hashCode());
          // just tagging this on here
          assertEquals(a.toString(), b.toString());
  
          a.setNullableZeroCopyByteBuffer(buf);
          b.setNullableZeroCopyByteBuffer(buf);
  
          assertEquals(a, b);
          assertEquals(a.hashCode(), b.hashCode());
          assertEquals(a.toString(), b.toString());
  
          a.setNullableZeroCopyByteBuffer(null);
          b.setNullableZeroCopyByteBuffer(null);
  
          assertEquals(a, b);
          assertEquals(a.hashCode(), b.hashCode());
          assertEquals(a.toString(), b.toString());
      }
  
      @Test
      public void testMyTaggedIntArray() {
          // Verify that the tagged int array reads as empty when not set.
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals(Collections.emptyList(), message.myTaggedIntArray()));
  
          // Verify that we can set a tagged array of ints.
          testRoundTrip(new SimpleExampleMessageData().
                  setMyTaggedIntArray(Arrays.asList(1, 2, 3)),
              message -> assertEquals(Arrays.asList(1, 2, 3), message.myTaggedIntArray()));
      }
  
      @Test
      public void testMyNullableString() {
          // Verify that the tagged field reads as null when not set.
          testRoundTrip(new SimpleExampleMessageData(), message -> assertNull(message.myNullableString()));
  
          // Verify that we can set and retrieve a string for the tagged field.
          testRoundTrip(new SimpleExampleMessageData().setMyNullableString("foobar"),
              message -> assertEquals("foobar", message.myNullableString()));
      }
  
      @Test
      public void testMyInt16() {
          // Verify that the tagged field reads as 123 when not set.
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals((short) 123, message.myInt16()));
  
          testRoundTrip(new SimpleExampleMessageData().setMyInt16((short) 456),
              message -> assertEquals((short) 456, message.myInt16()));
      }
  
      @Test
      public void testMyUint32() {
          // Verify that the uint16 field reads as 33000 when not set.
          testRoundTrip(new SimpleExampleMessageData(),
                  message -> assertEquals(1234567, message.myUint32()));
  
          testRoundTrip(new SimpleExampleMessageData().setMyUint32(123),
                  message -> assertEquals(123, message.myUint32()));
          testRoundTrip(new SimpleExampleMessageData().setMyUint32(60000),
                  message -> assertEquals(60000, message.myUint32()));
      }
  
      @Test
      public void testMyUint16() {
          // Verify that the uint16 field reads as 33000 when not set.
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals(33000, message.myUint16()));
  
          testRoundTrip(new SimpleExampleMessageData().setMyUint16(123),
              message -> assertEquals(123, message.myUint16()));
          testRoundTrip(new SimpleExampleMessageData().setMyUint16(60000),
              message -> assertEquals(60000, message.myUint16()));
      }
  
      @Test
      public void testMyString() {
          // Verify that the tagged field reads as empty when not set.
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals("", message.myString()));
  
          testRoundTrip(new SimpleExampleMessageData().setMyString("abc"),
              message -> assertEquals("abc", message.myString()));
      }
  
      @Test
      public void testMyBytes() {
          assertThrows(RuntimeException.class,
              () -> new SimpleExampleMessageData().setMyUint16(-1));
          assertThrows(RuntimeException.class,
              () -> new SimpleExampleMessageData().setMyUint16(UNSIGNED_SHORT_MAX + 1));
  
          assertThrows(RuntimeException.class,
                  () -> new SimpleExampleMessageData().setMyUint32(-1));
          assertThrows(RuntimeException.class,
                  () -> new SimpleExampleMessageData().setMyUint32(UNSIGNED_INT_MAX + 1));
  
          // Verify that the tagged field reads as empty when not set.
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertArrayEquals(new byte[0], message.myBytes()));
  
          testRoundTrip(new SimpleExampleMessageData().
                  setMyBytes(new byte[] {0x43, 0x66}),
              message -> assertArrayEquals(new byte[] {0x43, 0x66},
                  message.myBytes()));
  
          testRoundTrip(new SimpleExampleMessageData().setMyBytes(null), message -> assertNull(message.myBytes()));
      }
  
      @Test
      public void testTaggedUuid() {
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals(
                  Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"),
                  message.taggedUuid()));
  
          Uuid randomUuid = Uuid.randomUuid();
          testRoundTrip(new SimpleExampleMessageData().
                  setTaggedUuid(randomUuid),
              message -> assertEquals(
                  randomUuid,
                  message.taggedUuid()));
      }
  
      @Test
      public void testTaggedLong() {
          testRoundTrip(new SimpleExampleMessageData(),
              message -> assertEquals(0xcafcacafcacafcaL,
                  message.taggedLong()));
  
          testRoundTrip(new SimpleExampleMessageData().
                  setMyString("blah").
                  setMyTaggedIntArray(Collections.singletonList(4)).
                  setTaggedLong(0x123443211234432L),
              message -> assertEquals(0x123443211234432L,
                  message.taggedLong()));
      }
  
      @Test
      public void testMyStruct() {
          // Verify that we can set and retrieve a nullable struct object.
          SimpleExampleMessageData.MyStruct myStruct =
              new SimpleExampleMessageData.MyStruct().setStructId(10).setArrayInStruct(
                  Collections.singletonList(new SimpleExampleMessageData.StructArray().setArrayFieldId(20))
              );
          testRoundTrip(new SimpleExampleMessageData().setMyStruct(myStruct),
              message -> assertEquals(myStruct, message.myStruct()), (short) 2);
      }
  
      @Test
      public void testMyStructUnsupportedVersion() {
          SimpleExampleMessageData.MyStruct myStruct =
                  new SimpleExampleMessageData.MyStruct().setStructId(10);
          // Check serialization throws exception for unsupported version
          assertThrows(UnsupportedVersionException.class,
              () -> testRoundTrip(new SimpleExampleMessageData().setMyStruct(myStruct), (short) 1));
      }
  
      /**
       * Check following cases:
       * 1. Tagged struct can be serialized/deserialized for version it is supported
       * 2. Tagged struct doesn't matter for versions it is not declared.
       */
      @Test
      public void testMyTaggedStruct() {
          // Verify that we can set and retrieve a nullable struct object.
          SimpleExampleMessageData.TaggedStruct myStruct =
              new SimpleExampleMessageData.TaggedStruct().setStructId("abc");
          testRoundTrip(new SimpleExampleMessageData().setMyTaggedStruct(myStruct),
              message -> assertEquals(myStruct, message.myTaggedStruct()), (short) 2);
  
          // Not setting field works for both version 1 and version 2 protocol
          testRoundTrip(new SimpleExampleMessageData().setMyString("abc"),
              message -> assertEquals("abc", message.myString()), (short) 1);
          testRoundTrip(new SimpleExampleMessageData().setMyString("abc"),
              message -> assertEquals("abc", message.myString()), (short) 2);
      }
  
      @Test
      public void testCommonStruct() {
          SimpleExampleMessageData message = new SimpleExampleMessageData();
          message.setMyCommonStruct(new SimpleExampleMessageData.TestCommonStruct()
              .setFoo(1)
              .setBar(2));
          message.setMyOtherCommonStruct(new SimpleExampleMessageData.TestCommonStruct()
              .setFoo(3)
              .setBar(4));
          testRoundTrip(message, (short) 2);
      }
  
      private SimpleExampleMessageData deserialize(ByteBuffer buf, short version) {
          SimpleExampleMessageData message = new SimpleExampleMessageData();
          message.read(new ByteBufferAccessor(buf.duplicate()), version);
          return message;
      }
  
      private void testRoundTrip(SimpleExampleMessageData message, short version) {
          testRoundTrip(message, m -> { }, version);
      }
  
      private void testRoundTrip(SimpleExampleMessageData message,
                                 Consumer<SimpleExampleMessageData> validator) {
          testRoundTrip(message, validator, (short) 1);
      }
  
      private void testRoundTrip(SimpleExampleMessageData message,
                                 Consumer<SimpleExampleMessageData> validator,
                                 short version) {
          validator.accept(message);
  
          SimpleExampleMessageData message2 = roundTripSerde(message, version);
          validator.accept(message2);
          assertEquals(message, message2);
          assertEquals(message.hashCode(), message2.hashCode());
  
          // Check JSON serialization
          JsonNode serializedJson = SimpleExampleMessageDataJsonConverter.write(message, version);
          SimpleExampleMessageData messageFromJson = SimpleExampleMessageDataJsonConverter.read(serializedJson, version);
          validator.accept(messageFromJson);
          assertEquals(message, messageFromJson);
          assertEquals(message.hashCode(), messageFromJson.hashCode());
      }
  
      private SimpleExampleMessageData roundTripSerde(
          SimpleExampleMessageData message,
          short version
      ) {
          ByteBuffer buf = MessageUtil.toByteBufferAccessor(message, version).buffer();
          // Check size calculation
          assertEquals(buf.remaining(), message.size(new ObjectSerializationCache(), version));
          return deserialize(buf.duplicate(), version);
      }
  
      @Test
      public void testTaggedFieldsShouldSupportFlexibleVersionSubset() {
          SimpleExampleMessageData message = new SimpleExampleMessageData()
              .setTaggedLongFlexibleVersionSubset(15L);
  
          testRoundTrip(
              message,
              msg -> assertEquals(15, msg.taggedLongFlexibleVersionSubset),
              (short) 2
          );
  
          SimpleExampleMessageData deserialized = roundTripSerde(message, (short) 1);
          assertEquals(new SimpleExampleMessageData(), deserialized);
          assertEquals(0, deserialized.taggedLongFlexibleVersionSubset);
      }
  
      @Test
      public void testToString() {
          SimpleExampleMessageData message = new SimpleExampleMessageData();
          message.setMyUint16(65535);
          message.setTaggedUuid(Uuid.fromString("x7D3Ck_ZRA22-dzIvu_pnQ"));
          message.setMyFloat64(1.0);
          assertEquals("SimpleExampleMessageData(processId=AAAAAAAAAAAAAAAAAAAAAA, " +
                  "myTaggedIntArray=[], " +
                  "myNullableString=null, " +
                  "myInt16=123, myFloat64=1.0, " +
                  "myString='', " +
                  "myBytes=[], " +
                  "taggedUuid=x7D3Ck_ZRA22-dzIvu_pnQ, " +
                  "taggedLong=914172222550880202, " +
                  "zeroCopyByteBuffer=java.nio.HeapByteBuffer[pos=0 lim=0 cap=0], " +
                  "nullableZeroCopyByteBuffer=java.nio.HeapByteBuffer[pos=0 lim=0 cap=0], " +
                  "myStruct=MyStruct(structId=0, arrayInStruct=[]), " +
                  "myTaggedStruct=TaggedStruct(structId=''), " +
                  "taggedLongFlexibleVersionSubset=0, " +
                  "myCommonStruct=TestCommonStruct(foo=123, bar=123), " +
                  "myOtherCommonStruct=TestCommonStruct(foo=123, bar=123), " +
                  "myUint16=65535, " +
                  "myUint32=1234567)", message.toString());
      }
  }
  