/*
 * 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.ignite.internal.metastorage.client;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.ignite.internal.metastorage.common.OperationType;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageCommandListener;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.ClusterLocalConfiguration;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.ClusterServiceFactory;
import org.apache.ignite.network.internal.recovery.message.HandshakeStartMessage;
import org.apache.ignite.network.internal.recovery.message.HandshakeStartMessageSerializationFactory;
import org.apache.ignite.network.internal.recovery.message.HandshakeStartResponseMessage;
import org.apache.ignite.network.internal.recovery.message.HandshakeStartResponseMessageSerializationFactory;
import org.apache.ignite.network.scalecube.ScaleCubeClusterServiceFactory;
import org.apache.ignite.network.scalecube.message.ScaleCubeMessage;
import org.apache.ignite.network.scalecube.message.ScaleCubeMessageSerializationFactory;
import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.message.RaftClientMessageFactory;
import org.apache.ignite.raft.client.message.impl.RaftClientMessageFactoryImpl;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.client.service.impl.RaftGroupServiceImpl;
import org.apache.ignite.raft.server.RaftServer;
import org.apache.ignite.raft.server.impl.RaftServerImpl;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;

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;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

/**
 * Meta storage client tests.
 */
@SuppressWarnings("WeakerAccess")
public class ITMetaStorageServiceTest {
    /** The logger. */
    private static final IgniteLogger LOG = IgniteLogger.forClass(ITMetaStorageServiceTest.class);

    /** Base network port. */
    private static final int NODE_PORT_BASE = 20_000;

    /** Nodes. */
    private static final int NODES = 2;

    /** */
    private static final String METASTORAGE_RAFT_GROUP_NAME = "METASTORAGE_RAFT_GROUP";

    /** */
    public static final int LATEST_REVISION = -1;

    /** Factory. */
    private static final RaftClientMessageFactory FACTORY = new RaftClientMessageFactoryImpl();

    /** Network factory. */
    private static final ClusterServiceFactory NETWORK_FACTORY = new ScaleCubeClusterServiceFactory();

    /** */
    private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistry()
        .registerFactory(ScaleCubeMessage.TYPE, new ScaleCubeMessageSerializationFactory())
        .registerFactory(HandshakeStartMessage.TYPE, new HandshakeStartMessageSerializationFactory())
        .registerFactory(HandshakeStartResponseMessage.TYPE, new HandshakeStartResponseMessageSerializationFactory());

    /**  Expected server result entry. */
    private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY =
            new org.apache.ignite.internal.metastorage.server.Entry(
                    new byte[] {1},
                    new byte[] {2},
                    10,
                    2
            );

    /**  Expected server result entry. */
    private static final EntryImpl EXPECTED_RESULT_ENTRY =
            new EntryImpl(
                    new ByteArray(new byte[] {1}),
                    new byte[] {2},
                    10,
                    2
            );

    /** Expected result map. */
    private static final NavigableMap<ByteArray, Entry> EXPECTED_RESULT_MAP;

    private static final Collection<org.apache.ignite.internal.metastorage.server.Entry> EXPECTED_SRV_RESULT_COLL;

    /** Cluster. */
    private ArrayList<ClusterService> cluster = new ArrayList<>();

    /**  Meta storage raft server. */
    private RaftServer metaStorageRaftSrv;

    static {
        EXPECTED_RESULT_MAP = new TreeMap<>();

        EntryImpl entry1 = new EntryImpl(
                new ByteArray(new byte[]{1}),
                new byte[]{2},
                10,
                2
        );

        EXPECTED_RESULT_MAP.put(entry1.key(), entry1);

        EntryImpl entry2 = new EntryImpl(
                new ByteArray(new byte[]{3}),
                new byte[]{4},
                10,
                3
        );

        EXPECTED_RESULT_MAP.put(entry2.key(), entry2);

        EXPECTED_SRV_RESULT_COLL = new ArrayList<>();

        EXPECTED_SRV_RESULT_COLL.add(new org.apache.ignite.internal.metastorage.server.Entry(
                entry1.key().bytes(), entry1.value(), entry1.revision(), entry1.updateCounter()
        ));

        EXPECTED_SRV_RESULT_COLL.add(new org.apache.ignite.internal.metastorage.server.Entry(
                entry2.key().bytes(), entry2.value(), entry2.revision(), entry2.updateCounter()
        ));
    }

    /**
     * Run @{code} NODES cluster nodes.
     */
    @BeforeEach
    public void beforeTest() {
        for (int i = 0; i < NODES; i++) {
            cluster.add(
                    startClusterNode(
                            "node_" + i,
                            NODE_PORT_BASE + i,
                            IntStream.range(NODE_PORT_BASE, NODE_PORT_BASE + NODES).boxed().
                                    map((port) -> "localhost:" + port).collect(Collectors.toList())));
        }

        for (ClusterService node : cluster)
            assertTrue(waitForTopology(node, NODES, 1000));

        LOG.info("Cluster started.");
    }

    /**
     * Shutdown raft server and stop all cluster nodes.
     *
     * @throws Exception If failed to shutdown raft server,
     */
    @AfterEach
    public void afterTest() throws Exception {
        metaStorageRaftSrv.shutdown();

        for (ClusterService node : cluster)
            node.shutdown();
    }

    /**
     * Tests {@link MetaStorageService#get(ByteArray)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGet() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key) {
                        return EXPECTED_SRV_RESULT_ENTRY;
                    }
                });

        assertEquals(EXPECTED_RESULT_ENTRY, metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
    }

    /**
     * Tests {@link MetaStorageService#get(ByteArray, long)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetWithUpperBoundRevision() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key, long rev) {
                        return EXPECTED_SRV_RESULT_ENTRY;
                    }
                });

        assertEquals(
                EXPECTED_RESULT_ENTRY,
                metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key(), EXPECTED_RESULT_ENTRY.revision()).get()
        );
    }

    /**
     * Tests {@link MetaStorageService#getAll(Set)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAll() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull List<org.apache.ignite.internal.metastorage.server.Entry> getAll(List<byte[]> keys) {
                        return new ArrayList<>(EXPECTED_SRV_RESULT_COLL);
                    }
                });

        assertEquals(EXPECTED_RESULT_MAP, metaStorageSvc.getAll(EXPECTED_RESULT_MAP.keySet()).get());
    }

    /**
     * Tests {@link MetaStorageService#getAll(Set, long)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAllWithUpperBoundRevision() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull List<org.apache.ignite.internal.metastorage.server.Entry> getAll(List<byte[]> keys, long revUpperBound) {
                        return new ArrayList<>(EXPECTED_SRV_RESULT_COLL);
                    }
                });

        assertEquals(
                EXPECTED_RESULT_MAP,
                metaStorageSvc.getAll(EXPECTED_RESULT_MAP.keySet(), 10).get()
        );
    }

    /**
     * Tests {@link MetaStorageService#put(ByteArray, byte[])}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testPut() throws Exception {
        ByteArray expKey = new ByteArray(new byte[]{1});

        byte[] expVal = new byte[]{2};

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @SuppressWarnings("JavaAbbreviationUsage")
                    @Override public void put(byte[] key, byte[] value) {
                        assertArrayEquals(expKey.bytes(), key);

                        assertArrayEquals(expVal, value);
                    }
                });

        metaStorageSvc.put(expKey, expVal).get();
    }

    /**
     * Tests {@link MetaStorageService#getAndPut(ByteArray, byte[])}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAndPut() throws Exception {
        byte[] expVal = new byte[]{2};

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @SuppressWarnings("JavaAbbreviationUsage")
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry getAndPut(byte[] key, byte[] value) {
                        assertArrayEquals(EXPECTED_RESULT_ENTRY.key().bytes(), key);

                        assertArrayEquals(expVal, value);

                        return EXPECTED_SRV_RESULT_ENTRY;
                    }
                });

        assertEquals(
                EXPECTED_RESULT_ENTRY,
                metaStorageSvc.getAndPut(EXPECTED_RESULT_ENTRY.key(), expVal).get()
        );
    }

    /**
     * Tests {@link MetaStorageService#putAll(Map)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testPutAll() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public void putAll(List<byte[]> keys, List<byte[]> values) {
                        // Assert keys equality.
                        assertEquals(EXPECTED_RESULT_MAP.keySet().size(), keys.size());

                        List<byte[]> expKeys = EXPECTED_RESULT_MAP.keySet().stream().
                                map(ByteArray::bytes).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expKeys.get(i), keys.get(i));


                        // Assert values equality.
                        assertEquals(EXPECTED_RESULT_MAP.values().size(), values.size());

                        List<byte[]> expVals = EXPECTED_RESULT_MAP.values().stream().
                                map(Entry::value).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expVals.get(i), values.get(i));
                    }
                });

        metaStorageSvc.putAll(
                EXPECTED_RESULT_MAP.entrySet().stream()
                        .collect(Collectors.toMap(
                                Map.Entry::getKey,
                                e -> e.getValue().value())
                        )
        ).get();
    }

    /**
     * Tests {@link MetaStorageService#getAndPutAll(Map)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAndPutAll() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull List<org.apache.ignite.internal.metastorage.server.Entry> getAndPutAll(List<byte[]> keys, List<byte[]> values) {
                        // Assert keys equality.
                        assertEquals(EXPECTED_RESULT_MAP.keySet().size(), keys.size());

                        List<byte[]> expKeys = EXPECTED_RESULT_MAP.keySet().stream().
                                map(ByteArray::bytes).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expKeys.get(i), keys.get(i));

                        // Assert values equality.
                        assertEquals(EXPECTED_RESULT_MAP.values().size(), values.size());

                        List<byte[]> expVals = EXPECTED_RESULT_MAP.values().stream().
                                map(Entry::value).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expVals.get(i), values.get(i));

                        return new ArrayList<>(EXPECTED_SRV_RESULT_COLL);
                    }
                });

        Map<ByteArray, Entry> gotRes = metaStorageSvc.getAndPutAll(
                EXPECTED_RESULT_MAP.entrySet().stream()
                        .collect(Collectors.toMap(
                                Map.Entry::getKey,
                                e -> e.getValue().value())
                        )
        ).get();

        assertEquals(EXPECTED_RESULT_MAP, gotRes);
    }

    /**
     * Tests {@link MetaStorageService#remove(ByteArray)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRemove() throws Exception {
        ByteArray expKey = new ByteArray(new byte[]{1});

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public void remove(byte[] key) {
                        assertArrayEquals(expKey.bytes(), key);
                    }
                });

        metaStorageSvc.remove(expKey).get();
    }

    /**
     * Tests {@link MetaStorageService#getAndRemove(ByteArray)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAndRemove() throws Exception {
        EntryImpl expRes = new EntryImpl(
                new ByteArray(new byte[]{1}),
                new byte[]{3},
                10,
                2
        );

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry getAndRemove(byte[] key) {
                        assertArrayEquals(expRes.key().bytes(), key);

                        return new org.apache.ignite.internal.metastorage.server.Entry(
                                expRes.key().bytes(),
                                expRes.value(),
                                expRes.revision(),
                                expRes.updateCounter()
                        );
                    }
                });

        assertEquals(expRes, metaStorageSvc.getAndRemove(expRes.key()).get());
    }

    /**
     * Tests {@link MetaStorageService#removeAll(Set)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRemoveAll() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public void removeAll(List<byte[]> keys) {
                        assertEquals(EXPECTED_RESULT_MAP.keySet().size(), keys.size());

                        List<byte[]> expKeys = EXPECTED_RESULT_MAP.keySet().stream().
                                map(ByteArray::bytes).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expKeys.get(i), keys.get(i));
                    }
                });

        metaStorageSvc.removeAll(EXPECTED_RESULT_MAP.keySet()).get();
    }

    /**
     * Tests {@link MetaStorageService#getAndRemoveAll(Set)}.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testGetAndRemoveAll() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull List<org.apache.ignite.internal.metastorage.server.Entry> getAndRemoveAll(List<byte[]> keys) {
                        // Assert keys equality.
                        assertEquals(EXPECTED_RESULT_MAP.keySet().size(), keys.size());

                        List<byte[]> expKeys = EXPECTED_RESULT_MAP.keySet().stream().
                                map(ByteArray::bytes).collect(Collectors.toList());

                        for (int i = 0; i < expKeys.size(); i++)
                            assertArrayEquals(expKeys.get(i), keys.get(i));

                        return new ArrayList<>(EXPECTED_SRV_RESULT_COLL);
                    }
                });

        Map<ByteArray, Entry> gotRes = metaStorageSvc.getAndRemoveAll(EXPECTED_RESULT_MAP.keySet()).get();

        assertEquals(EXPECTED_RESULT_MAP, gotRes);
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with not null keyTo and explicit revUpperBound.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeWitKeyToAndUpperBound() throws Exception {
        ByteArray expKeyFrom = new ByteArray(new byte[]{1});

        ByteArray expKeyTo = new ByteArray(new byte[]{3});

        long expRevUpperBound = 10;

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        assertArrayEquals(expKeyFrom.bytes(), keyFrom);

                        assertArrayEquals(expKeyTo.bytes(), keyTo);

                        assertEquals(expRevUpperBound, revUpperBound);

                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.Entry> it = new Iterator<>() {
                                @Override public boolean hasNext() {
                                    return false;
                                }

                                @Override public org.apache.ignite.internal.metastorage.server.Entry next() {
                                    return null;
                                }
                            };


                            @Override public void close() throws Exception {

                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.Entry> iterator() {
                                return it;
                            }

                            @Override
                            public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public org.apache.ignite.internal.metastorage.server.Entry next() {
                                return it.next();
                            }
                        };
                    }
                });

        metaStorageSvc.range(expKeyFrom, expKeyTo, expRevUpperBound).close();
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with not null keyTo.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeWitKeyTo() throws Exception {
        ByteArray expKeyFrom = new ByteArray(new byte[]{1});

        ByteArray expKeyTo = new ByteArray(new byte[]{3});

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        assertArrayEquals(expKeyFrom.bytes(), keyFrom);

                        assertArrayEquals(expKeyTo.bytes(), keyTo);

                        assertEquals(LATEST_REVISION, revUpperBound);

                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.Entry> it = new Iterator<>() {
                                @Override
                                public boolean hasNext() {
                                    return false;
                                }

                                @Override
                                public org.apache.ignite.internal.metastorage.server.Entry next() {
                                    return null;
                                }
                            };

                            @Override public void close() throws Exception {

                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.Entry> iterator() {
                                return it;
                            }

                            @Override public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override public org.apache.ignite.internal.metastorage.server.Entry next() {
                                return it.next();
                            }
                        };
                    }
                });

        metaStorageSvc.range(expKeyFrom, expKeyTo).close();
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with null keyTo.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeWitNullAsKeyTo() throws Exception {
        ByteArray expKeyFrom = new ByteArray(new byte[]{1});

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        assertArrayEquals(expKeyFrom.bytes(), keyFrom);

                        assertNull(keyTo);

                        assertEquals(LATEST_REVISION, revUpperBound);

                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.Entry> it = new Iterator<>() {
                                @Override public boolean hasNext() {
                                    return false;
                                }

                                @Override public org.apache.ignite.internal.metastorage.server.Entry next() {
                                    return null;
                                }
                            };

                            @Override public void close() throws Exception {

                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.Entry> iterator() {
                                return it;
                            }

                            @Override public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public org.apache.ignite.internal.metastorage.server.Entry next() {
                                return it.next();
                            }
                        };
                    }
                });

        metaStorageSvc.range(expKeyFrom, null).close();
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} hasNext.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeHasNext() throws Exception {
        ByteArray expKeyFrom = new ByteArray(new byte[]{1});

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.Entry> it = new Iterator<>() {
                                @Override public boolean hasNext() {
                                    return true;
                                }

                                @Override public org.apache.ignite.internal.metastorage.server.Entry next() {
                                    return null;
                                }
                            };

                            @Override public void close() throws Exception {

                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.Entry> iterator() {
                                return it;
                            }

                            @Override public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public org.apache.ignite.internal.metastorage.server.Entry next() {
                                return it.next();
                            }
                        };
                    }
                });

        Cursor<Entry> cursor = metaStorageSvc.range(expKeyFrom, null);

        assertTrue(cursor.iterator().hasNext());
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} next.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeNext() throws Exception {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.Entry> it = new Iterator<>() {
                                @Override public boolean hasNext() {
                                    return true;
                                }

                                @Override public org.apache.ignite.internal.metastorage.server.Entry next() {
                                    return EXPECTED_SRV_RESULT_ENTRY;
                                }
                            };

                            @Override public void close() throws Exception {

                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.Entry> iterator() {
                                return it;
                            }

                            @Override public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public org.apache.ignite.internal.metastorage.server.Entry next() {
                                return it.next();
                            }
                        };
                    }
                });

        Cursor<Entry> cursor = metaStorageSvc.range(EXPECTED_RESULT_ENTRY.key(), null);

        assertEquals(EXPECTED_RESULT_ENTRY, (cursor.iterator().next()));
    }

    /**
     * Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} close.
     *
     * @throws Exception If failed.
     */
    @Test
    public void testRangeClose() throws Exception {
        ByteArray expKeyFrom = new ByteArray(new byte[]{1});

        Cursor cursorMock = mock(Cursor.class);

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
                        return cursorMock;
                    }
                });

        Cursor<Entry> cursor = metaStorageSvc.range(expKeyFrom, null);

        cursor.close();

        verify(cursorMock, times(1)).close();
    }

    @Test
    public void testWatchOnUpdate() throws Exception {
        org.apache.ignite.internal.metastorage.server.WatchEvent returnedWatchEvents = new org.apache.ignite.internal.metastorage.server.WatchEvent(List.of(
                new org.apache.ignite.internal.metastorage.server.EntryEvent(
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[]{2},
                                new byte[]{20},
                                1,
                                1
                        ),
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[]{2},
                                new byte[]{21},
                                2,
                                4
                        )
                ),
                new org.apache.ignite.internal.metastorage.server.EntryEvent(
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[] {3},
                                new byte[] {20},
                                1,
                                2
                        ),
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[] {3},
                                new byte[]{},
                                2,
                                5
                        )
                ),
                new org.apache.ignite.internal.metastorage.server.EntryEvent(
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[] {4},
                                new byte[] {20},
                                1,
                                3
                        ),
                        new org.apache.ignite.internal.metastorage.server.Entry(
                                new byte[] {4},
                                new byte[] {},
                                3,
                                6
                        )
                )
        ));

        ByteArray keyFrom = new ByteArray(new byte[]{1});

        ByteArray keyTo = new ByteArray(new byte[]{10});

        long rev = 2;

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public Cursor<org.apache.ignite.internal.metastorage.server.WatchEvent> watch(byte[] keyFrom, byte[] keyTo, long rev) {
                        return new Cursor<>() {
                            private final Iterator<org.apache.ignite.internal.metastorage.server.WatchEvent> it = new Iterator<>() {
                                @Override public boolean hasNext() {

                                    return retirevedItemCnt.get() < returnedWatchEvents.entryEvents().size();
                                }

                                @Override public org.apache.ignite.internal.metastorage.server.WatchEvent next() {
                                    return returnedWatchEvents;
                                }
                            };

                            AtomicInteger retirevedItemCnt = new AtomicInteger(0);

                            @Override public void close() throws Exception {
                                // No-op.
                            }

                            @NotNull @Override public Iterator<org.apache.ignite.internal.metastorage.server.WatchEvent> iterator() {
                                return it;
                            }

                            @Override public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public org.apache.ignite.internal.metastorage.server.WatchEvent next() {
                                return it.next();
                            }
                        };
                    }
                });

        CountDownLatch latch = new CountDownLatch(1);

        IgniteUuid watchId = metaStorageSvc.watch(keyFrom, keyTo, rev, new WatchListener() {
            @Override public boolean onUpdate(@NotNull WatchEvent events) {
                List gotEvents = new ArrayList();

                List returnedWatchEvents = new ArrayList(events.entryEvents());

                Iterator<EntryEvent> iter = events.entryEvents().iterator();

                while (iter.hasNext())
                    gotEvents.add(iter.next());

                assertEquals(3, gotEvents.size());

                assertTrue(gotEvents.contains(returnedWatchEvents.get(0)));

                assertTrue(gotEvents.contains(returnedWatchEvents.get(1)));

                latch.countDown();

                return true;
            }

            @Override public void onError(@NotNull Throwable e) {
                // Within given test it's not expected to get here.
                fail();
            }
        }).get();

        latch.await();

        metaStorageSvc.stopWatch(watchId).get();
    }

    @Test
    public void testInvoke() throws Exception {
        ByteArray expKey = new ByteArray(new byte[]{1});

        byte[] expVal = new byte[]{2};

        Condition condition = Conditions.notExists(expKey);

        Operation success = Operations.put(expKey, expVal);

        Operation failure = Operations.noop();

        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public boolean invoke(
                            org.apache.ignite.internal.metastorage.server.Condition cond,
                            Collection<org.apache.ignite.internal.metastorage.server.Operation> success,
                            Collection<org.apache.ignite.internal.metastorage.server.Operation> failure) {
                        assertArrayEquals(expKey.bytes(), cond.key());

                        assertArrayEquals(expKey.bytes(), success.iterator().next().key());
                        assertArrayEquals(expVal, success.iterator().next().value());

                        assertEquals(OperationType.NO_OP, failure.iterator().next().type());

                        return true;
                    }
                });

        assertTrue(metaStorageSvc.invoke(condition, success, failure).get());
    }

    // TODO: IGNITE-14693 Add tests for exception handling logic: onError,
    // TODO: (CompactedException | OperationTimeoutException)

    /**
     * Tests {@link MetaStorageService#get(ByteArray)}.
     *
     * @throws Exception If failed.
     */
    @Disabled // TODO: IGNITE-14693 Add tests for exception handling logic.
    @Test
    public void testGetThatThrowsCompactedException() {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key) {
                        throw new org.apache.ignite.internal.metastorage.server.CompactedException();
                    }
                });

        assertThrows(CompactedException.class, () -> metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
    }

    /**
     * Tests {@link MetaStorageService#get(ByteArray)}.
     *
     * @throws Exception If failed.
     */
    @Disabled // TODO: IGNITE-14693 Add tests for exception handling logic.
    @Test
    public void testGetThatThrowsOperationTimeoutException() {
        MetaStorageService metaStorageSvc = prepareMetaStorage(
                new AbstractKeyValueStorage() {
                    @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key) {
                        throw new OperationTimeoutException();
                    }
                });

        assertThrows(OperationTimeoutException.class, () -> metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
    }

    /**
     * @param name Node name.
     * @param port Local port.
     * @param srvs Server nodes of the cluster.
     * @return The client cluster view.
     */
    private ClusterService startClusterNode(String name, int port, List<String> srvs) {
        var ctx = new ClusterLocalConfiguration(name, port, srvs, SERIALIZATION_REGISTRY);

        var net = NETWORK_FACTORY.createClusterService(ctx);

        net.start();

        return net;
    }

    /**
     * @param cluster The cluster.
     * @param exp Expected count.
     * @param timeout The timeout in millis.
     * @return {@code True} if topology size is equal to expected.
     */
    @SuppressWarnings("SameParameterValue")
    private boolean waitForTopology(ClusterService cluster, int exp, int timeout) {
        long stop = System.currentTimeMillis() + timeout;

        while (System.currentTimeMillis() < stop) {
            if (cluster.topologyService().allMembers().size() >= exp)
                return true;

            try {
                Thread.sleep(50);
            }
            catch (InterruptedException e) {
                return false;
            }
        }

        return false;
    }

    /**
     * Prepares meta storage by instantiating corresponding raft server with MetaStorageCommandListener and
     * {@link MetaStorageServiceImpl}.
     *
     * @param keyValStorageMock {@link KeyValueStorage} mock.
     * @return {@link MetaStorageService} instance.
     */
    private MetaStorageService prepareMetaStorage(KeyValueStorage keyValStorageMock) {
        metaStorageRaftSrv = new RaftServerImpl(
                cluster.get(0),
                FACTORY,
                1000,
                Map.of(METASTORAGE_RAFT_GROUP_NAME, new MetaStorageCommandListener(keyValStorageMock))
        );

        RaftGroupService metaStorageRaftGrpSvc = new RaftGroupServiceImpl(
                METASTORAGE_RAFT_GROUP_NAME,
                cluster.get(1),
                FACTORY,
                10_000,
                List.of(new Peer(cluster.get(0).topologyService().localMember())),
                true,
                200
        );

        return new MetaStorageServiceImpl(metaStorageRaftGrpSvc);
    }

    /**
     * Abstract {@link KeyValueStorage}. Used for tests purposes.
     */
    @SuppressWarnings("JavaAbbreviationUsage")
    private abstract class AbstractKeyValueStorage implements KeyValueStorage {
        /** {@inheritDoc} */
        @Override public long revision() {
            return 0;
        }

        /** {@inheritDoc} */
        @Override public long updateCounter() {
            return 0;
        }

        /** {@inheritDoc} */
        @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry get(byte[] key, long rev) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public @NotNull Collection<org.apache.ignite.internal.metastorage.server.Entry> getAll(List<byte[]> keys) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public @NotNull Collection<org.apache.ignite.internal.metastorage.server.Entry> getAll(List<byte[]> keys, long revUpperBound) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void put(byte[] key, byte[] value) {
            fail();
        }

        /** {@inheritDoc} */
        @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry getAndPut(byte[] key, byte[] value) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void putAll(List<byte[]> keys, List<byte[]> values) {
            fail();
        }

        /** {@inheritDoc} */
        @Override public @NotNull Collection<org.apache.ignite.internal.metastorage.server.Entry> getAndPutAll(List<byte[]> keys, List<byte[]> values) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void remove(byte[] key) {
            fail();
        }

        /** {@inheritDoc} */
        @Override public @NotNull org.apache.ignite.internal.metastorage.server.Entry getAndRemove(byte[] key) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void removeAll(List<byte[]> keys) {
            fail();
        }

        /** {@inheritDoc} */
        @Override public @NotNull Collection<org.apache.ignite.internal.metastorage.server.Entry> getAndRemoveAll(List<byte[]> keys) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public boolean invoke(
                org.apache.ignite.internal.metastorage.server.Condition condition,
                Collection<org.apache.ignite.internal.metastorage.server.Operation> success,
                Collection<org.apache.ignite.internal.metastorage.server.Operation> failure
        ) {
            fail();

            return false;
        }

        /** {@inheritDoc} */
        @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public Cursor<org.apache.ignite.internal.metastorage.server.Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public Cursor<org.apache.ignite.internal.metastorage.server.WatchEvent> watch(byte[] keyFrom, byte[] keyTo, long rev) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public Cursor<org.apache.ignite.internal.metastorage.server.WatchEvent> watch(byte[] key, long rev) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public Cursor<org.apache.ignite.internal.metastorage.server.WatchEvent> watch(Collection<byte[]> keys, long rev) {
            fail();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void compact() {
            fail();
        }
    }
}