modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectMetadataExchangeMultinodeTest.java

/*
 * 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.processors.cache.binary;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.binary.BinaryObjectBuilder;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.PA;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.GridTestUtils.DiscoveryHook;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;

/**
 *
 */
public class GridCacheBinaryObjectMetadataExchangeMultinodeTest extends GridCommonAbstractTest {
    /** */
    private boolean applyDiscoveryHook;

    /** */
    private DiscoveryHook discoveryHook;

    /** */
    private static final String BINARY_TYPE_NAME = "TestBinaryType";

    /** */
    private static final int BINARY_TYPE_ID = 708045005;

    /** */
    private static final long MAX_AWAIT = 9_000;

    /** */
    private static final AtomicInteger metadataReqsCounter = new AtomicInteger(0);

    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
        IgniteConfiguration cfg = super.getConfiguration(gridName);

        if (applyDiscoveryHook && discoveryHook != null)
            ((TestTcpDiscoverySpi)cfg.getDiscoverySpi()).discoveryHook(discoveryHook);

        cfg.setMarshaller(new BinaryMarshaller());

        CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);

        ccfg.setCacheMode(CacheMode.REPLICATED);
        ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);

        cfg.setCacheConfiguration(ccfg);

        return cfg;
    }

    /** {@inheritDoc} */
    @Override protected void afterTest() throws Exception {
        super.afterTest();

        stopAllGrids();
    }

    /**
     * Verifies that if thread tries to read metadata with ongoing update it gets blocked
     * until acknowledge message arrives.
     */
    @Test
    public void testReadRequestBlockedOnUpdatingMetadata() throws Exception {
        final CyclicBarrier barrier = new CyclicBarrier(2);

        applyDiscoveryHook = false;

        final Ignite ignite0 = startGrid(0);
        final Ignite ignite1 = startGrid(1);

        final GridFutureAdapter finishFut = new GridFutureAdapter();

        applyDiscoveryHook = true;
        discoveryHook = new DiscoveryHook() {
            private volatile IgniteEx ignite;

            @Override public void beforeDiscovery(DiscoveryCustomMessage customMsg) {
                if (finishFut.isDone())
                    return;

                if (customMsg instanceof MetadataUpdateAcceptedMessage) {
                    MetadataUpdateAcceptedMessage acceptedMsg = (MetadataUpdateAcceptedMessage)customMsg;
                    if (acceptedMsg.typeId() == BINARY_TYPE_ID && acceptedMsg.acceptedVersion() == 2) {
                        Object binaryProc = U.field(ignite.context(), "cacheObjProc");
                        Object transport = U.field(binaryProc, "transport");

                        try {
                            barrier.await(MAX_AWAIT, TimeUnit.MILLISECONDS);

                            Map syncMap = U.field(transport, "syncMap");

                            GridTestUtils.waitForCondition(new PA() {
                                @Override public boolean apply() {
                                    return syncMap.size() == 1;
                                }
                            }, MAX_AWAIT);

                            assertEquals("unexpected size of syncMap: ", 1, syncMap.size());

                            Object syncKey = syncMap.keySet().iterator().next();

                            int typeId = U.field(syncKey, "typeId");
                            assertEquals("unexpected typeId: ", BINARY_TYPE_ID, typeId);

                            int ver = U.field(syncKey, "ver");
                            assertEquals("unexpected pendingVersion: ", 2, ver);

                            finishFut.onDone();
                        }
                        catch (Throwable t) {
                            finishFut.onDone(t);
                        }
                    }
                }
            }

            @Override public void ignite(IgniteEx ignite) {
                this.ignite = ignite;
            }
        };

        final IgniteEx ignite2 = startGrid(2);
        discoveryHook.ignite(ignite2);

        // Unfinished PME may affect max await timeout.
        awaitPartitionMapExchange();

        // Update metadata (version 1).
        ignite0.executorService(ignite0.cluster().forLocal()).submit(new Runnable() {
            @Override public void run() {
                addIntField(ignite0, "f1", 101, 1);
            }
        }).get();

        // Update metadata (version 2).
        ignite1.executorService(ignite1.cluster().forLocal()).submit(new Runnable() {
            @Override public void run() {
                addStringField(ignite1, "f2", "str", 2);
            }
        });

        // Read metadata.
        IgniteFuture readFut = ignite2.compute(ignite2.cluster().forLocal()).callAsync(new IgniteCallable<Object>() {
            @Override public Object call() throws Exception {
                barrier.await(MAX_AWAIT, TimeUnit.MILLISECONDS);

                return ((BinaryObject)ignite2.cache(DEFAULT_CACHE_NAME).withKeepBinary().get(1)).field("f1");
            }
        });

        finishFut.get(MAX_AWAIT);

        assertEquals(101, readFut.get(MAX_AWAIT));
    }

    /**
     * Verifies that all sequential updates that don't introduce any conflicts are accepted and observed by all nodes.
     */
    @Test
    public void testSequentialUpdatesNoConflicts() throws Exception {
        IgniteEx ignite0 = startGrid(0);

        final IgniteEx ignite1 = startGrid(1);

        final String intFieldName = "f1";

        ignite1.executorService().submit(new Runnable() {
            @Override public void run() {
                addIntField(ignite1, intFieldName, 101, 1);
            }
        }).get();

        int fld = ((BinaryObject)ignite0.cache(DEFAULT_CACHE_NAME).withKeepBinary().get(1)).field(intFieldName);

        assertEquals(fld, 101);

        final IgniteEx ignite2 = startGrid(2);

        final String strFieldName = "f2";

        ignite2.executorService().submit(new Runnable() {
            @Override public void run() {
                addStringField(ignite2, strFieldName, "str", 2);
            }
        }).get();

        assertEquals(((BinaryObject)ignite1.cache(DEFAULT_CACHE_NAME).withKeepBinary().get(2)).field(strFieldName), "str");
    }

    /**
     * Verifies that client is able to detect obsolete metadata situation and request up-to-date from the cluster.
     */
    @Test
    public void testClientRequestsUpToDateMetadata() throws Exception {
        final IgniteEx ignite0 = startGrid(0);

        final IgniteEx ignite1 = startGrid(1);

        ignite0.executorService().submit(new Runnable() {
            @Override public void run() {
                addIntField(ignite0, "f1", 101, 1);
            }
        }).get();

        final Ignite client = startDeafClient("client");

        ClusterGroup clientGrp = client.cluster().forClients();

        final String strVal = "strVal101";

        ignite1.executorService().submit(new Runnable() {
            @Override public void run() {
                addStringField(ignite1, "f2", strVal, 1);
            }
        }).get();

        String res = client.compute(clientGrp).call(new IgniteCallable<String>() {
            @Override public String call() throws Exception {
                return ((BinaryObject)client.cache(DEFAULT_CACHE_NAME).withKeepBinary().get(1)).field("f2");
            }
        });

        assertEquals(strVal, res);
    }

    /**
     * Verifies that client resends request for up-to-date metadata in case of failure on server received first request.
     */
    @Test
    public void testClientRequestsUpToDateMetadataOneNodeDies() throws Exception {
        final Ignite srv0 = startGrid(0);
        replaceWithStoppingMappingRequestListener(((GridKernalContext)U.field(srv0, "ctx")).io(), 0);

        final Ignite srv1 = startGrid(1);
        replaceWithCountingMappingRequestListener(((GridKernalContext)U.field(srv1, "ctx")).io());

        final Ignite srv2 = startGrid(2);
        replaceWithCountingMappingRequestListener(((GridKernalContext)U.field(srv2, "ctx")).io());

        final Ignite client = startDeafClient("client");

        ClusterGroup clientGrp = client.cluster().forClients();

        srv0.executorService().submit(new Runnable() {
            @Override public void run() {
                addStringField(srv0, "f2", "strVal101", 0);
            }
        }).get();

        client.compute(clientGrp).call(new IgniteCallable<String>() {
            @Override public String call() throws Exception {
                return ((BinaryObject)client.cache(DEFAULT_CACHE_NAME).withKeepBinary().get(0)).field("f2");
            }
        });

        assertEquals(metadataReqsCounter.get(), 2);
    }

    /**
     * Starts client node that skips <b>MetadataUpdateProposedMessage</b> and <b>MetadataUpdateAcceptedMessage</b>
     * messages.
     *
     * @param clientName name of client node.
     */
    private Ignite startDeafClient(String clientName) throws Exception {
        applyDiscoveryHook = true;
        discoveryHook = new DiscoveryHook() {
            @Override public void beforeDiscovery(DiscoveryCustomMessage customMsg) {
                if (customMsg instanceof MetadataUpdateProposedMessage) {
                    if (((MetadataUpdateProposedMessage)customMsg).typeId() == BINARY_TYPE_ID)
                        GridTestUtils.setFieldValue(customMsg, "typeId", 1);
                }
                else if (customMsg instanceof MetadataUpdateAcceptedMessage) {
                    if (((MetadataUpdateAcceptedMessage)customMsg).typeId() == BINARY_TYPE_ID)
                        GridTestUtils.setFieldValue(customMsg, "typeId", 1);
                }
            }
        };

        Ignite client = startClientGrid(clientName);

        applyDiscoveryHook = false;

        return client;
    }

    /**
     *
     */
    private void replaceWithStoppingMappingRequestListener(GridIoManager ioMgr, final int nodeIdToStop) {
        ioMgr.removeMessageListener(GridTopic.TOPIC_METADATA_REQ);

        ioMgr.addMessageListener(GridTopic.TOPIC_METADATA_REQ, new GridMessageListener() {
            @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
                new Thread(new Runnable() {
                    @Override public void run() {
                        metadataReqsCounter.incrementAndGet();
                        stopGrid(nodeIdToStop, true);
                    }
                }).start();
            }
        });
    }

    /**
     *
     */
    private void replaceWithCountingMappingRequestListener(GridIoManager ioMgr) {
        GridMessageListener[] lsnrs = U.field(ioMgr, "sysLsnrs");

        final GridMessageListener delegate = lsnrs[GridTopic.TOPIC_METADATA_REQ.ordinal()];

        GridMessageListener wrapper = new GridMessageListener() {
            @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
                metadataReqsCounter.incrementAndGet();
                delegate.onMessage(nodeId, msg, plc);
            }
        };

        lsnrs[GridTopic.TOPIC_METADATA_REQ.ordinal()] = wrapper;
    }

    /**
     * Adds field of integer type to fixed binary type.
     *
     * @param ignite Ignite.
     * @param fieldName Field name.
     * @param fieldVal Field value.
     * @param cacheIdx Cache index.
     */
    private void addIntField(Ignite ignite, String fieldName, int fieldVal, int cacheIdx) {
        BinaryObjectBuilder builder = ignite.binary().builder(BINARY_TYPE_NAME);

        IgniteCache<Object, Object> cache = ignite.cache(DEFAULT_CACHE_NAME).withKeepBinary();

        builder.setField(fieldName, fieldVal);

        cache.put(cacheIdx, builder.build());
    }

    /**
     * Adds field of String type to fixed binary type.
     *
     * @param ignite Ignite.
     * @param fieldName Field name.
     * @param fieldVal Field value.
     * @param cacheIdx Cache index.
     */
    private void addStringField(Ignite ignite, String fieldName, String fieldVal, int cacheIdx) {
        BinaryObjectBuilder builder = ignite.binary().builder(BINARY_TYPE_NAME);

        IgniteCache<Object, Object> cache = ignite.cache(DEFAULT_CACHE_NAME).withKeepBinary();

        builder.setField(fieldName, fieldVal);

        cache.put(cacheIdx, builder.build());
    }
}