Avoid excessive garbage in map#clear() for TS [HZ-3620]

hazelcast/src/main/java/com/hazelcast/map/impl/operation/ClearOperation.java

@@ -21,6 +21,7 @@
 import com.hazelcast.map.impl.operation.steps.ClearOpSteps;
 import com.hazelcast.map.impl.operation.steps.engine.State;
 import com.hazelcast.map.impl.operation.steps.engine.Step;
+import com.hazelcast.map.impl.recordstore.DefaultRecordStore;
 import com.hazelcast.spi.impl.operationservice.BackupAwareOperation;
 import com.hazelcast.spi.impl.operationservice.MutatingOperation;
 import com.hazelcast.spi.impl.operationservice.Operation;

hazelcast/src/main/java/com/hazelcast/map/impl/operation/steps/ClearOpSteps.java

@@ -22,9 +22,12 @@
 import com.hazelcast.map.impl.operation.steps.engine.Step;
 import com.hazelcast.map.impl.record.Record;
 import com.hazelcast.map.impl.recordstore.DefaultRecordStore;
+import com.hazelcast.map.impl.recordstore.RecordStore;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.BiConsumer;
 
@@ -45,21 +48,32 @@ public void runStep(State state) {
             ArrayList<Data> keys = new ArrayList<>();
             ArrayList<Record> records = new ArrayList<>();
             // we don't remove locked keys. These are clearable records.
-            recordStore.forEach(new BiConsumer<>() {
-                final Set<Data> lockedKeySet = recordStore.getLockStore().getLockedKeys();
-
-                @Override
-                public void accept(Data dataKey, Record record) {
-                    if (lockedKeySet != null && !lockedKeySet.contains(dataKey)) {
-                        keys.add(recordStore.isTieredStorageEnabled() ? toHeapData(dataKey) : dataKey);
+            final Set<Data> lockedKeySet = recordStore.getLockStore().getLockedKeys();
+            final Iterator<Map.Entry<Data, Record>> iterator = recordStore.iterator();
+            boolean tieredStorageEnabled = recordStore.isTieredStorageEnabled();;
+
+            while (iterator.hasNext()) {
+                Map.Entry<Data, Record> entry = iterator.next();
+                Data dataKey = entry.getKey();
+                Record record = entry.getValue();
+
+                if (lockedKeySet != null && !lockedKeySet.contains(dataKey)) {
+                    keys.add(tieredStorageEnabled ? toHeapData(dataKey) : dataKey);
+                    if (!recordStore.isTieredStorageEnabled()) {
                         records.add(record);
                     }
+                }
 
+                if (keys.size() == BATCH_SIZE) {
+                    // Batch filling is completed
+                    break;
                 }
-            }, false);
+            }
 
             state.setKeys(keys);
-            state.setRecords(records);
+            if (!tieredStorageEnabled) {
+                state.setRecords(records);
+            }
         }
 
         @Override
@@ -102,6 +116,17 @@ public void runStep(State state) {
 
         @Override
         public Step nextStep(State state) {
+            RecordStore recordStore = state.getRecordStore();
+            int currentSize = recordStore.size();
+            final Set<Data> lockedKeySet = ((DefaultRecordStore) recordStore).getLockStore().getLockedKeys();
+            int lockedSize = lockedKeySet.size();
+
+            if (currentSize - lockedSize > 0) {
+                // We still have entries to process
+                // Process them in the next batch
+                return CLEAR_MEMORY;
+            }
+
             return UtilSteps.FINAL_STEP;
         }
     };

hazelcast/src/main/java/com/hazelcast/map/impl/operation/steps/IMapOpStep.java

@@ -29,6 +29,11 @@
  */
 public interface IMapOpStep extends Step<State> {
 
+    /**
+     * The batch size of records handled in bulk operations.
+     */
+    int BATCH_SIZE = 1000;
+
     /**
      * Decides when to offload based on configured map-store type.
      * <p>

hazelcast/src/main/java/com/hazelcast/map/impl/recordstore/DefaultRecordStore.java

@@ -46,6 +46,8 @@
 import com.hazelcast.map.impl.mapstore.writebehind.WriteBehindStore;
 import com.hazelcast.map.impl.mapstore.writebehind.entry.DelayedEntry;
 import com.hazelcast.map.impl.operation.MapOperation;
+import com.hazelcast.map.impl.operation.steps.ClearOpSteps;
+import com.hazelcast.map.impl.operation.steps.engine.Step;
 import com.hazelcast.map.impl.querycache.QueryCacheContext;
 import com.hazelcast.map.impl.querycache.publisher.MapPublisherRegistry;
 import com.hazelcast.map.impl.querycache.publisher.PublisherContext;
@@ -112,6 +114,12 @@ public class DefaultRecordStore extends AbstractEvictableRecordStore {
      */
     protected final Collection<Future<?>> loadingFutures = new ConcurrentLinkedQueue<>();
 
+    /**
+     * Defined by {@link com.hazelcast.spi.properties.ClusterProperty#WAN_REPLICATE_IMAP_EVICTIONS},
+     * if set to true then eviction operations by this RecordStore will be WAN replicated
+     */
+    protected boolean wanReplicateEvictions;
+
     /**
      * A reference to the Json Metadata store. It is initialized lazily only if the
      * store is needed.
@@ -131,11 +139,6 @@ public class DefaultRecordStore extends AbstractEvictableRecordStore {
      * key loading.
      */
     private boolean loadedOnPreMigration;
-    /**
-     * Defined by {@link com.hazelcast.spi.properties.ClusterProperty#WAN_REPLICATE_IMAP_EVICTIONS},
-     * if set to true then eviction operations by this RecordStore will be WAN replicated
-     */
-    private boolean wanReplicateEvictions;
 
     private final IPartitionService partitionService;
     private final InterceptorRegistry interceptorRegistry;