/*
 * 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.coordinator.group;

import org.apache.kafka.common.protocol.ApiMessage;
import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.Deserializer;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataValue;
import org.apache.kafka.coordinator.group.generated.CoordinatorRecordType;
import org.apache.kafka.server.common.ApiMessageAndVersion;

import org.junit.jupiter.api.Test;

import java.nio.ByteBuffer;

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;
import static org.junit.jupiter.api.Assertions.assertTrue;

@SuppressWarnings("ClassDataAbstractionCoupling")
public class GroupCoordinatorRecordSerdeTest {
    @Test
    public void testSerializeKey() {
        GroupCoordinatorRecordSerde serializer = new GroupCoordinatorRecordSerde();
        CoordinatorRecord record = CoordinatorRecord.record(
            new ConsumerGroupMetadataKey().setGroupId("group"),
            new ApiMessageAndVersion(
                new ConsumerGroupMetadataValue().setEpoch(10),
                (short) 0
            )
        );

        assertArrayEquals(
            MessageUtil.toVersionPrefixedBytes(record.key().apiKey(), record.key()),
            serializer.serializeKey(record)
        );
    }

    @Test
    public void testSerializeValue() {
        GroupCoordinatorRecordSerde serializer = new GroupCoordinatorRecordSerde();
        CoordinatorRecord record = CoordinatorRecord.record(
            new ConsumerGroupMetadataKey().setGroupId("group"),
            new ApiMessageAndVersion(
                new ConsumerGroupMetadataValue().setEpoch(10),
                (short) 0
            )
        );

        assertArrayEquals(
            MessageUtil.toVersionPrefixedBytes(record.value().version(), record.value().message()),
            serializer.serializeValue(record)
        );
    }

    @Test
    public void testSerializeNullValue() {
        GroupCoordinatorRecordSerde serializer = new GroupCoordinatorRecordSerde();
        CoordinatorRecord record = CoordinatorRecord.tombstone(
            new ConsumerGroupMetadataKey().setGroupId("group")
        );

        assertNull(serializer.serializeValue(record));
    }

    @Test
    public void testDeserialize() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ApiMessage key = new ConsumerGroupMetadataKey().setGroupId("foo");
        ByteBuffer keyBuffer = MessageUtil.toCoordinatorTypePrefixedByteBuffer(key);

        ApiMessageAndVersion value = new ApiMessageAndVersion(
            new ConsumerGroupMetadataValue().setEpoch(10),
            (short) 0
        );
        ByteBuffer valueBuffer = MessageUtil.toVersionPrefixedByteBuffer(value.version(), value.message());

        CoordinatorRecord record = serde.deserialize(keyBuffer, valueBuffer);
        assertEquals(key, record.key());
        assertEquals(value, record.value());
    }

    @Test
    public void testDeserializeWithTombstoneForValue() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ApiMessage key = new ConsumerGroupMetadataKey().setGroupId("foo");
        ByteBuffer keyBuffer = MessageUtil.toCoordinatorTypePrefixedByteBuffer(key);

        CoordinatorRecord record = serde.deserialize(keyBuffer, null);
        assertEquals(key, record.key());
        assertNull(record.value());
    }

    @Test
    public void testDeserializeWithInvalidRecordType() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ByteBuffer keyBuffer = ByteBuffer.allocate(64);
        keyBuffer.putShort((short) 255);
        keyBuffer.rewind();

        ByteBuffer valueBuffer = ByteBuffer.allocate(64);

        Deserializer.UnknownRecordTypeException ex =
            assertThrows(Deserializer.UnknownRecordTypeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertEquals((short) 255, ex.unknownType());
    }

    @Test
    public void testDeserializeWithKeyEmptyBuffer() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ByteBuffer keyBuffer = ByteBuffer.allocate(0);
        ByteBuffer valueBuffer = ByteBuffer.allocate(64);

        RuntimeException ex =
            assertThrows(RuntimeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertEquals("Could not read version from key's buffer.", ex.getMessage());
    }

    @Test
    public void testDeserializeWithValueEmptyBuffer() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ApiMessageAndVersion key = new ApiMessageAndVersion(
            new ConsumerGroupMetadataKey().setGroupId("foo"),
            (short) 3
        );
        ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());

        ByteBuffer valueBuffer = ByteBuffer.allocate(0);

        RuntimeException ex =
            assertThrows(RuntimeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertEquals("Could not read version from value's buffer.", ex.getMessage());
    }

    @Test
    public void testDeserializeWithInvalidKeyBytes() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ByteBuffer keyBuffer = ByteBuffer.allocate(2);
        keyBuffer.putShort((short) 3);
        keyBuffer.rewind();

        ByteBuffer valueBuffer = ByteBuffer.allocate(2);
        valueBuffer.putShort((short) 0);
        valueBuffer.rewind();

        RuntimeException ex =
            assertThrows(RuntimeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertTrue(ex.getMessage().startsWith("Could not read record with version 3 from key's buffer due to"),
            ex.getMessage());
    }

    @Test
    public void testDeserializeWithInvalidValueBytes() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ApiMessageAndVersion key = new ApiMessageAndVersion(
            new ConsumerGroupMetadataKey().setGroupId("foo"),
            (short) 3
        );
        ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());

        ByteBuffer valueBuffer = ByteBuffer.allocate(2);
        valueBuffer.putShort((short) 0);
        valueBuffer.rewind();

        RuntimeException ex =
            assertThrows(RuntimeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from value's buffer due to"),
            ex.getMessage());
    }

    @Test
    public void testDeserializeWithInvalidValueVersion() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ApiMessage key = new ConsumerGroupMetadataKey().setGroupId("foo");
        ByteBuffer keyBuffer = MessageUtil.toCoordinatorTypePrefixedByteBuffer(key);

        ByteBuffer valueBuffer1 = ByteBuffer.allocate(2);
        valueBuffer1.putShort((short) (ConsumerGroupMetadataValue.HIGHEST_SUPPORTED_VERSION + 1));
        valueBuffer1.rewind();

        Deserializer.UnknownRecordVersionException ex =
            assertThrows(Deserializer.UnknownRecordVersionException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer1));
        assertEquals(key.apiKey(), ex.type());
        assertEquals(ConsumerGroupMetadataValue.HIGHEST_SUPPORTED_VERSION + 1, ex.unknownVersion());

        keyBuffer.rewind();
        ByteBuffer valueBuffer2 = ByteBuffer.allocate(2);
        valueBuffer2.putShort((short) (ConsumerGroupMetadataValue.LOWEST_SUPPORTED_VERSION - 1));
        valueBuffer2.rewind();

        ex = assertThrows(Deserializer.UnknownRecordVersionException.class,
            () -> serde.deserialize(keyBuffer, valueBuffer2));
        assertEquals(key.apiKey(), ex.type());
        assertEquals(ConsumerGroupMetadataValue.LOWEST_SUPPORTED_VERSION - 1, ex.unknownVersion());
    }

    @Test
    public void testDeserializeAllRecordTypes() {
        for (CoordinatorRecordType record : CoordinatorRecordType.values()) {
            roundTrip(record.newRecordKey(), record.newRecordValue());
        }
    }

    private void roundTrip(
        ApiMessage key,
        ApiMessage val
    ) {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        for (short version = val.lowestSupportedVersion(); version < val.highestSupportedVersion(); version++) {
            ApiMessageAndVersion valMessageAndVersion = new ApiMessageAndVersion(val, version);

            CoordinatorRecord record = serde.deserialize(
                MessageUtil.toCoordinatorTypePrefixedByteBuffer(key),
                MessageUtil.toVersionPrefixedByteBuffer(version, val)
            );

            assertEquals(key, record.key());
            assertEquals(valMessageAndVersion, record.value());
        }
    }
}
