@Override
    public Object add(Data key, Object value,
                      long expirationTime, long now, UUID transactionId) {
        // When using format InMemoryFormat.NATIVE,
        // just copy key & value to heap.
        if (NATIVE == inMemoryFormat) {
            value = toHeapData(value);
            key = toHeapData(key);
        }

        // This note describes the problem when we want to persist
        // all states of an entry (means write-coalescing is off)
        // by using both EntryProcessor + OBJECT in-memory-format:
        //
        // If in-memory-format is OBJECT, there is a possibility that a previous
        // state of an entry can be overwritten by a subsequent write operation
        // while both are waiting in the write-behind-queue, this is because
        // they are referencing to the same entry-value. To prevent such a
        // problem, we are taking snapshot of the value by serializing it, this
        // means an extra serialization and additional latency for operations
        // like map#put, but it is needed, otherwise we can lose state.
        if (!coalesce && OBJECT == inMemoryFormat) {
            value = toHeapData(value);
        }
        expirationTime = getUserExpirationTime(expirationTime);
        add(DelayedEntries.newAddedDelayedEntry(key, value, expirationTime, now, partitionId, transactionId));
        return value;
    }
@Test
    public void shouldNotChangeSequenceAfterAddingToQueue() throws IllegalAccessException {
        // given
        when(mapStoreContext.getMapServiceContext().getNodeEngine().getSerializationService()).thenReturn(mock(InternalSerializationService.class));
        WriteBehindStore store = new WriteBehindStore(mapStoreContext, 1, writeBehindProcessor);
        DummyQueue queue = mock(DummyQueue.class, CALLS_REAL_METHODS);
        ReflectionUtils.setFieldValueReflectively(store, "writeBehindQueue", queue);
        DelayedEntry<Data, Object> delayedEntry = DelayedEntries.newAddedDelayedEntry(mock(HeapData.class), new Entry(1, 1), 0, 0, 1, null);
        assertEquals(0L, delayedEntry.getSequence());

        // when
        store.add(delayedEntry);

        // then
        assertEquals(1L, delayedEntry.getSequence());
        assertEquals(1L, queue.getSequence());

    }