Big objects bug fix.

Author: xtern

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java

@@ -756,7 +756,7 @@ else if (PageIdUtils.tag(pageId) != PageIdAllocator.FLAG_DATA)
                 int written;
 
                 if (dataRows.size() == 1) {
-                    written = fragment ? row.size() - (rows.size() % maxPayloadSize) : 0;
+                    written = fragment ? row.size() - (row.size() % maxPayloadSize) : 0;
 
                     written = write(pageId, writeRows, initIo, row, written, FAIL_I, statHolder);
                 } else

modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListPreloadWithBatchUpdatesTest.java

@@ -41,6 +41,7 @@
 import org.apache.ignite.internal.util.typedef.PA;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.After;
 import org.junit.Before;
@@ -62,7 +63,10 @@ public class FreeListPreloadWithBatchUpdatesTest extends GridCommonAbstractTest
     private static final int HDR_SIZE = 8 + 32;
 
     /** */
-    private static final long DEF_REG_SIZE = 6 * 1024 * 1024 * 1024L;
+    private static final long DEF_REG_SIZE_INIT = 3400 * 1024 * 1024L;
+
+    /** */
+    private static final long DEF_REG_SIZE = 6144 * 1024 * 1024L;
 
     /** */
     private static final String DEF_CACHE_NAME = "some-cache";
@@ -73,15 +77,15 @@ public static Iterable<Object[]> setup() {
         return Arrays.asList(new Object[][]{
             {CacheAtomicityMode.ATOMIC, false},
             {CacheAtomicityMode.ATOMIC, true},
-//            {CacheAtomicityMode.TRANSACTIONAL, false},
-//            {CacheAtomicityMode.TRANSACTIONAL, true},
-//            {CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT, false},
-//            {CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT, true}
+            {CacheAtomicityMode.TRANSACTIONAL, false},
+            {CacheAtomicityMode.TRANSACTIONAL, true},
+            {CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT, false},
+            {CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT, true}
         });
     }
 
     /** */
-    @Parameterized.Parameter(0)
+    @Parameterized.Parameter()
     public CacheAtomicityMode cacheAtomicityMode;
 
     /** */
@@ -93,7 +97,7 @@ public static Iterable<Object[]> setup() {
         IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
 
         DataRegionConfiguration def = new DataRegionConfiguration();
-        def.setInitialSize(3400 * 1024 * 1024L);
+        def.setInitialSize(DEF_REG_SIZE_INIT);
         def.setMaxSize(DEF_REG_SIZE);
         def.setPersistenceEnabled(persistence);
 
@@ -117,8 +121,6 @@ public static Iterable<Object[]> setup() {
     @Before
     public void before() throws Exception {
         cleanPersistenceDir();
-
-        System.setProperty(IGNITE_DATA_STORAGE_BATCH_PAGE_WRITE, "true");
     }
 
     /**
@@ -129,15 +131,13 @@ public void after() throws Exception {
         stopAllGrids();
 
         cleanPersistenceDir();
-
-        System.clearProperty(IGNITE_PDS_WAL_REBALANCE_THRESHOLD);
-        System.clearProperty(IGNITE_DATA_STORAGE_BATCH_PAGE_WRITE);
     }
 
     /**
      *
      */
     @Test
+    @WithSystemProperty(key = IGNITE_DATA_STORAGE_BATCH_PAGE_WRITE, value = "true")
     public void testBatchRebalance() throws Exception {
         Ignite node = startGrid(0);
 
@@ -148,16 +148,14 @@ public void testBatchRebalance() throws Exception {
         node.createCache(ccfg());
 
         int cnt = 100_000;
-        int minSize = 0;
-        int maxSize = 2048;
-        int start = 0;
 
-        log.info("Loading " + cnt + " random entries per " + minSize + " - " + maxSize + " bytes.");
+        int minSize = 0;
+        int maxSize = 16384;
 
         Map<Integer, byte[]> srcMap = new HashMap<>();
 
-        for (int i = start; i < start + cnt; i++) {
-            int size = minSize + ThreadLocalRandom.current().nextInt(maxSize - minSize);
+        for (int i = 0; i < cnt; i++) {
+            int size = maxSize == minSize ? maxSize : minSize + ThreadLocalRandom.current().nextInt(maxSize - minSize);
 
             byte[] obj = new byte[size];
 
@@ -168,13 +166,7 @@ public void testBatchRebalance() throws Exception {
             streamer.addData(srcMap);
         }
 
-        srcMap.put(1, new byte[65536]);
-
-        node.cache(DEF_CACHE_NAME).put(1, new byte[65536]);
-
-        log.info("Done");
-
-        IgniteCache cache = node.cache(DEF_CACHE_NAME);
+        log.info("Data loaded.");
 
         if (persistence)
             node.cluster().active(false);
@@ -191,16 +183,12 @@ public void testBatchRebalance() throws Exception {
             node.cluster().setBaselineTopology(list);
         }
 
-        log.info("await rebalance");
+        log.info("Await rebalance.");
 
         awaitRebalance(node2, DEF_CACHE_NAME);
 
-        U.sleep(2_000);
-
         node.close();
 
-        log.info("Verification on node2");
-
         validateCacheEntries(node2.cache(DEF_CACHE_NAME), srcMap);
 
         if (persistence) {
@@ -220,17 +208,17 @@ public void testBatchRebalance() throws Exception {
      *
      */
     @Test
+    @WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "100")
+    @WithSystemProperty(key = IGNITE_DATA_STORAGE_BATCH_PAGE_WRITE, value = "true")
     public void testBatchHistoricalRebalance() throws Exception {
         if (!persistence)
             return;
 
         // TODO https://issues.apache.org/jira/browse/IGNITE-7384
-        // http://apache-ignite-developers.2346864.n4.nabble.com/Historical-rebalance-td38380.html
+        // TODO http://apache-ignite-developers.2346864.n4.nabble.com/Historical-rebalance-td38380.html
         if (cacheAtomicityMode == CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT)
             return;
 
-        System.setProperty(IGNITE_PDS_WAL_REBALANCE_THRESHOLD, "100");
-
         Ignite node = startGrids(2);
 
         node.cluster().active(true);
@@ -297,6 +285,7 @@ public void testBatchHistoricalRebalance() throws Exception {
     /** */
     @Test
     @Ignore
+    @WithSystemProperty(key = IGNITE_DATA_STORAGE_BATCH_PAGE_WRITE, value = "true")
     public void checkStreamer() throws Exception {
         Ignite node = startGrids(4);
 
@@ -342,7 +331,7 @@ private void awaitRebalance(IgniteEx node, String name) throws IgniteInterrupted
             }
         }, 60_000);
 
-        U.sleep(1000);
+        U.sleep(3000);
 
         assertTrue(ok);
     }
@@ -353,20 +342,16 @@ private void awaitRebalance(IgniteEx node, String name) throws IgniteInterrupted
      */
     @SuppressWarnings("unchecked")
     private void validateCacheEntries(IgniteCache cache, Map<?, byte[]> map) {
-        log.info("Cache validation: " + map.size());
+        int size = cache.size();
 
-        assertEquals(map.size(), cache.size());
-
-        for (Map.Entry<?, byte[]> e : map.entrySet()) {
-            String idx = "idx=" + e.getKey();
+        assertEquals("Cache size mismatch.", map.size(), size);
 
-            byte[] bytes = (byte[])cache.get(e.getKey());
+        log.info("Validation " + cache.getName() + ", size=" + size);
 
-            assertNotNull(idx, bytes);
-
-            assertEquals(idx + ": length not equal", e.getValue().length, bytes.length);
+        for (Map.Entry<?, byte[]> e : map.entrySet()) {
+            String idx = "key=" + e.getKey();
 
-            assertArrayEquals(idx, e.getValue(), bytes);
+            assertArrayEquals(idx, e.getValue(), (byte[])cache.get(e.getKey()));
         }
     }
 
@@ -387,3 +372,5 @@ private <K, V> CacheConfiguration<K, V> ccfg(int parts, CacheMode mode) {
             .setAtomicityMode(cacheAtomicityMode);
     }
 }
+
+