/*
 * 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.geode.internal.cache.partitioned.rebalance;

import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import junitparams.Parameters;
import junitparams.naming.TestCaseName;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;

import org.apache.geode.DataSerializer;
import org.apache.geode.cache.partition.PartitionMemberInfo;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.partitioned.OfflineMemberDetails;
import org.apache.geode.internal.cache.partitioned.OfflineMemberDetailsImpl;
import org.apache.geode.internal.cache.partitioned.PRLoad;
import org.apache.geode.internal.cache.partitioned.PartitionMemberInfoImpl;
import org.apache.geode.internal.cache.partitioned.rebalance.BucketOperator.Completion;
import org.apache.geode.internal.cache.partitioned.rebalance.model.AddressComparor;
import org.apache.geode.internal.cache.partitioned.rebalance.model.Bucket;
import org.apache.geode.internal.cache.partitioned.rebalance.model.Member;
import org.apache.geode.internal.cache.partitioned.rebalance.model.PartitionedRegionLoadModel;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;

@RunWith(GeodeParamsRunner.class)
public class PartitionedRegionLoadModelJUnitTest {

  private static final int MAX_MOVES = 5000;
  private static final boolean DEBUG = true;
  private static final long MB = 1024 * 1024;
  private MyBucketOperator bucketOperator;
  private final PartitionedRegion partitionedRegion = mock(PartitionedRegion.class);
  final ClusterDistributionManager clusterDistributionManager =
      mock(ClusterDistributionManager.class);

  @Before
  public void setUp() {
    bucketOperator = new MyBucketOperator();

    when(partitionedRegion.getDistributionManager()).thenReturn(clusterDistributionManager);

  }

  /**
   * This test checks basic redundancy satisfaction. It creates two buckets with low redundancy and
   * 1 bucket with full redundancy and excepts copies of the low redundancy buckets to be made.
   */
  @Test
  public void testRedundancySatisfaction() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 2, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {1, 1, 1, 0}, new long[] {1, 1, 1, 0});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 1, 0, 1}, new long[] {0, 0, 0, 1});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);

    Set<PartitionMemberInfo> details = model.getPartitionedMemberDetails("a");
    assertThat(details).hasSize(2);

    PartitionMemberInfo[] memberInfo = details.toArray(new PartitionMemberInfo[0]);

    assertThat(memberInfo[0].getBucketCount()).isEqualTo(3);
    assertThat(memberInfo[0].getConfiguredMaxMemory()).isEqualTo(500);
    assertThat(memberInfo[0].getPrimaryCount()).isEqualTo(3);

    // we expect three moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(3);

    PartitionMemberInfo[] memberInfo2 =
        model.getPartitionedMemberDetails("a").toArray(new PartitionMemberInfo[0]);

    assertThat(memberInfo2[0].getBucketCount()).isEqualTo(4);
    assertThat(memberInfo2[0].getConfiguredMaxMemory()).isEqualTo(500);
    assertThat(memberInfo2[0].getPrimaryCount()).isEqualTo(3);

    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member2, 2));
    expectedCreates.add(new Create(member1, 3));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
  }

  /**
   * This test creates buckets with low redundancy, but only 1 of the buckets is small enough to be
   * copied. The other bucket should be rejected because it is too big..
   */
  @Test
  public void testRedundancySatisfactionWithSizeLimit() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 2, 3,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    // A member with 1 bucket with low redundancy, but it is too big to copy anywhere
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 50, 50, new long[] {30, 0, 0}, new long[] {1, 0, 0});
    // A member with 2 buckets with low redundancy that can be copied
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 40, 40, new long[] {0, 10, 10}, new long[] {0, 1, 1});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);

    // we expect 2 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(2);

    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member1, 1));
    expectedCreates.add(new Create(member1, 2));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
  }

  @Test
  public void testRedundancySatisfactionWithCriticalMember() throws UnknownHostException {
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 2, 3,
        getAddressComparor(false), Collections.singleton(member1), partitionedRegion);

    // this member has critical heap
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 50, 50, new long[] {10, 0, 0}, new long[] {1, 0, 0});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 40, 40, new long[] {0, 10, 10}, new long[] {0, 1, 1});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);

    // we expect 2 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(1);

    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member2, 0));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
  }

  /**
   * This test makes sure we ignore the size limit if requested
   */
  @Test
  public void testRedundancySatisfactionDoNotEnforceLocalMaxMemory() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 2, 3,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    // A member with 1 bucket with low redundancy, but it is too big to copy anywhere
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 50, 50, new long[] {30, 0, 0}, new long[] {1, 0, 0});
    // A member with 2 buckets with low redundancy that can be copied
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 40, 40, new long[] {0, 10, 10}, new long[] {0, 1, 1});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), false);

    // we expect 2 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(3);

    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member1, 1));
    expectedCreates.add(new Create(member1, 2));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
    Set<PartitionMemberInfo> afterDetails = model.getPartitionedMemberDetails("a");
    assertThat(afterDetails.size()).isEqualTo(2);
    for (PartitionMemberInfo member : afterDetails) {
      if (member.getDistributedMember().equals(member1)) {
        assertThat(member.getConfiguredMaxMemory()).isEqualTo(details1.getConfiguredMaxMemory());
      } else {
        assertThat(member.getConfiguredMaxMemory()).isEqualTo(details2.getConfiguredMaxMemory());
      }
    }
  }

  /**
   * Tests to make sure that redundancy satisfaction prefers to make redundant copies on members
   * with remote IP addresses.
   */
  @Test
  public void testRedundancySatisfactionPreferRemoteIp() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 3,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(LocalHostUtil.getLocalHost(), 3);

    // Create some buckets with low redundancy on members 1 and 2
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {30, 0, 0}, new long[] {1, 0, 0});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 30, 30}, new long[] {0, 1, 1});
    PartitionMemberInfoImpl details3 =
        buildDetails(member3, new long[] {0, 0, 0}, new long[] {0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);

    // we expect 3 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(3);

    // The buckets should all be created on member 3 because
    // it has a different ip address
    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member3, 0));
    expectedCreates.add(new Create(member3, 1));
    expectedCreates.add(new Create(member3, 2));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
  }

  @Test
  public void testRedundancySatisfactionEnforceRemoteIp() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 3,
        getAddressComparor(true), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    // Create some buckets with low redundancy on members 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {30, 30, 30}, new long[] {1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 0, 0}, new long[] {0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);

    // we expect 0 moves, because we're enforcing that we can't create
    // copies on the same IP.
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(0);

    assertThat(bucketOperator.creates).isEqualTo(Collections.emptyList());
  }

  @Test
  public void testMoveBucketsEnforceRemoteIp() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 0, 3,
        getAddressComparor(true), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    // Create some buckets with low redundancy on members 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {30, 30, 30}, new long[] {1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 0, 0}, new long[] {0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);

    // we expect 0 moves, because we're enforcing that we can't create
    // copies on the same IP.
    assertThat(doMoves(new CompositeDirector(true, true, true, true), model)).isEqualTo(1);

    List<Move> expectedMoves = new ArrayList<>();
    expectedMoves.add(new Move(member1, member2));
    assertThat(bucketOperator.bucketMoves).isEqualTo(expectedMoves);
  }

  /**
   * Tests to make sure that redundancy satisfaction balances between nodes to ensure an even load.
   */
  @Test
  public void testRedundancySatisfactionBalanced() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 3);
    // Create some buckets with low redundancy on member 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {1, 1, 1, 1}, new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    PartitionMemberInfoImpl details3 =
        buildDetails(member3, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);

    // we expect 4 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(4);

    // The bucket creates should alternate between members
    // so that the load is balanced.
    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member3, 1));
    expectedCreates.add(new Create(member2, 2));
    expectedCreates.add(new Create(member3, 3));
    assertThat(bucketOperator.creates).isEqualTo(expectedCreates);
  }

  /**
   * Tests to make sure that redundancy satisfaction balances between nodes to ensure an even load.
   */
  @Test
  public void testColocatedRegions() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 12,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 3);
    // Create some buckets with low redundancy on member 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1},
            new long[] {1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 250, 250, new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
            new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0});
    PartitionMemberInfoImpl details3 =
        buildDetails(member3, 250, 250, new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
            new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);
    model.addRegion("b", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);

    // we expect 4 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, true), model)).isEqualTo(18);

    // The bucket creates should alternate between members
    // so that the load is balanced.
    Set<Create> expectedCreates = new HashSet<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member3, 1));
    expectedCreates.add(new Create(member2, 2));
    expectedCreates.add(new Create(member3, 3));
    expectedCreates.add(new Create(member2, 4));
    expectedCreates.add(new Create(member3, 5));
    expectedCreates.add(new Create(member2, 6));
    expectedCreates.add(new Create(member3, 7));
    expectedCreates.add(new Create(member2, 8));
    expectedCreates.add(new Create(member3, 9));
    expectedCreates.add(new Create(member2, 10));
    expectedCreates.add(new Create(member3, 11));
    assertThat(new HashSet<>(bucketOperator.creates)).isEqualTo(expectedCreates);

    Set<Move> expectedMoves = new HashSet<>();
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member3));
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member3));
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member3));
    assertThat(new HashSet<>(bucketOperator.primaryMoves)).isEqualTo(expectedMoves);
  }

  @Test
  public void testIncompleteColocation() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 3);
    // Create some buckets with low redundancy on member 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {1, 1, 1, 1}, new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    PartitionMemberInfoImpl details3 =
        buildDetails(member3, new long[] {1, 1, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);
    model.addRegion("b", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);
    model.addRegion("c", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);
    // Add a region which is not created on all of the members.

    // Member 3 should not be considered a target for any moves.

    assertThat(doMoves(new CompositeDirector(true, true, false, true), model)).isEqualTo(6);

    // Everything should be creatd on member2
    Set<Create> expectedCreates = new HashSet<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member2, 1));
    expectedCreates.add(new Create(member2, 2));
    expectedCreates.add(new Create(member2, 3));
    assertThat(new HashSet<>(bucketOperator.creates)).isEqualTo(expectedCreates);

    Set<Move> expectedMoves = new HashSet<>();
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member2));
    assertThat(new HashSet<>(bucketOperator.primaryMoves)).isEqualTo(expectedMoves);
  }

  /**
   * Test that we enforce local max memory on a per region basis IE if one of the regions has a low
   * lmm, it will prevent a bucket move
   */
  @Test
  public void testColocationTwoNonEvictionRegionsEnforceLocalMaxMemory()
      throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);

    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    // Create some buckets with low redundancy on member 1 and enough lmm for region a
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 500, 500 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 500, 500 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);


    // Region b has a lmm of 2MB, so member2 should only accept 2 buckets
    PartitionMemberInfoImpl bDetails1 =
        buildDetails(member1, 2, 2 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl bDetails2 =
        buildDetails(member2, 2, 2 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("b", Arrays.asList(bDetails1, bDetails2), new FakeOfflineDetails(), true);


    assertThat(doMoves(new CompositeDirector(true, true, false, true), model)).isEqualTo(4);

    // Only (2+2)MB data should be create on member2
    Set<Create> expectedCreates = new HashSet<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member2, 1));
    assertThat(new HashSet<>(bucketOperator.creates)).isEqualTo(expectedCreates);

    Set<Move> expectedMoves = new HashSet<>();
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member2));
    assertThat(new HashSet<>(bucketOperator.primaryMoves)).isEqualTo(expectedMoves);
  }

  /**
   * Test that a region with enforceLocalMaxMemory disabled colocated with
   * a region with memory full and enforceLocalmaxMemory enabled will prevent a bucket move.
   */
  @Test
  public void testColocationOneNonEvictionRegionReachesLocalMaxMemoryLimit()
      throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 1, 8 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 1, 8 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), false);


    PartitionMemberInfoImpl bDetails1 =
        buildDetails(member1, 1, 2 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl bDetails2 =
        buildDetails(member2, 1, 2 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("b", Arrays.asList(bDetails1, bDetails2), new FakeOfflineDetails(), true);


    assertThat(doMoves(new CompositeDirector(true, true, false, true), model)).isEqualTo(4);

    Set<Create> expectedCreates = new HashSet<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member2, 1));
    assertThat(new HashSet<>(bucketOperator.creates)).isEqualTo(expectedCreates);

    Set<Move> expectedMoves = new HashSet<>();
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member2));
    assertThat(new HashSet<>(bucketOperator.primaryMoves)).isEqualTo(expectedMoves);
  }

  /**
   * Test that a region with memory full and enforceLocalMaxMemory disabled will not prevent a
   * bucket move.
   */
  @Test
  public void testColocationOneEvictionRegionReachesLocalMaxMemoryLimit()
      throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);

    PartitionMemberInfoImpl details1 =
        buildDetails(member1, 1, 4 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, 1, 4 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2), new FakeOfflineDetails(), true);


    PartitionMemberInfoImpl bDetails1 =
        buildDetails(member1, 1, 2 * MB, new long[] {1 * MB, 1 * MB, 1 * MB, 1 * MB},
            new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl bDetails2 =
        buildDetails(member2, 1, 2 * MB, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("b", Arrays.asList(bDetails1, bDetails2), new FakeOfflineDetails(), false);


    assertThat(doMoves(new CompositeDirector(true, true, false, true), model)).isEqualTo(6);

    // Everything should be created on member2
    Set<Create> expectedCreates = new HashSet<>();
    expectedCreates.add(new Create(member2, 0));
    expectedCreates.add(new Create(member2, 1));
    expectedCreates.add(new Create(member2, 2));
    expectedCreates.add(new Create(member2, 3));
    assertThat(new HashSet<>(bucketOperator.creates)).isEqualTo(expectedCreates);

    Set<Move> expectedMoves = new HashSet<>();
    expectedMoves.add(new Move(member1, member2));
    expectedMoves.add(new Move(member1, member2));
    assertThat(new HashSet<>(bucketOperator.primaryMoves)).isEqualTo(expectedMoves);
  }

  /**
   * Test which illustrates the problem with our greedy algorithm. It doesn't necessarily end up
   * with a balanced result.
   *
   * TODO rebalance - change this test or fix the algorithm?
   */
  @Ignore
  @Test
  public void testFoolGreedyAlgorithm() throws UnknownHostException {
    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(bucketOperator, 1, 50,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 3);
    PartitionMemberInfoImpl details1 = buildDetails(member1,
        new long[] {1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
            0, 0, 0, 0},
        new long[] {1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
            0, 0, 0, 0});
    PartitionMemberInfoImpl details2 = buildDetails(member2,
        new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
            1, 1, 1, 1},
        new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
            1, 1, 1, 1});
    PartitionMemberInfoImpl details3 = buildDetails(member3,
        new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
            0, 0, 0},
        new long[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
            0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);

    doMoves(new CompositeDirector(true, true, false, false), model);

    // we'd like to have 20 buckets per member, but what we'll find is that member 1
    // will have 15 and 2 and 3 will have 17 and 18.
    for (PartitionMemberInfo details : model.getPartitionedMemberDetails("a")) {
      assertThat(details.getBucketCount()).isEqualTo(20);
    }
  }

  /**
   * Tests to make sure that redundancy satisfaction balances between nodes to ensure an even load.
   */
  @Test
  public void testRedundancySatisfactionWithFailures() throws UnknownHostException {
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    final InternalDistributedMember member2 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 2);
    InternalDistributedMember member3 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 3);

    MyBucketOperator op = new MyBucketOperator() {
      @Override
      public void createRedundantBucket(InternalDistributedMember targetMember, int i,
          Map<String, Long> colocatedRegionBytes, Completion completion) {
        if (targetMember.equals(member2)) {
          completion.onFailure();
        } else {
          super.createRedundantBucket(targetMember, i, colocatedRegionBytes, completion);
        }
      }
    };

    PartitionedRegionLoadModel model = new PartitionedRegionLoadModel(op, 1, 4,
        getAddressComparor(false), Collections.emptySet(), partitionedRegion);

    // Create some buckets with low redundancy on member 1
    PartitionMemberInfoImpl details1 =
        buildDetails(member1, new long[] {1, 1, 1, 1}, new long[] {1, 1, 1, 1});
    PartitionMemberInfoImpl details2 =
        buildDetails(member2, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    PartitionMemberInfoImpl details3 =
        buildDetails(member3, new long[] {0, 0, 0, 0}, new long[] {0, 0, 0, 0});
    model.addRegion("a", Arrays.asList(details1, details2, details3), new FakeOfflineDetails(),
        true);

    // we expect 8 moves
    assertThat(doMoves(new CompositeDirector(true, true, false, false), model)).isEqualTo(8);

    // The bucket creates should do all of the creates on member 3
    // because member2 failed.
    List<Create> expectedCreates = new ArrayList<>();
    expectedCreates.add(new Create(member3, 0));
    expectedCreates.add(new Create(member3, 1));
    expectedCreates.add(new Create(member3, 2));
    expectedCreates.add(new Create(member3, 3));
    assertThat(op.creates).isEqualTo(expectedCreates);
  }

  /**
   * Test that redundancy satisfation can handle asynchronous failures and complete the job
   * correctly.
   */
  @Test
  public void testRedundancySatisfactionWithAsyncFailures() throws UnknownHostException {
    InternalDistributedMember member1 =
        new InternalDistributedMember(InetAddress.getByName("127.0.0.1"), 1);
    InternalDistributedMember member2 =
