@Test
  public void testS2S2CDeltaPropagationWithHAOverflow() {
    VM clientVM = getController();
    VM server1VM = vm0;
    VM server2VM = vm1;

    for (VM vm : toArray(clientVM, server1VM, server2VM)) {
      vm.invoke(this::prepareDeltas);
    }

    for (VM vm : toArray(server1VM, server2VM)) {
      vm.invoke(() -> {
        System.setProperty(INTERNAL_FACTORY_PROPERTY, CACHE_CLIENT_PROXY_FACTORY_NAME);
        LATCH.set(new CountDownLatch(1));
      });
    }

    server1VM.invoke(() -> new ServerFactory()
        .regionShortcut(RegionShortcut.REPLICATE)
        .create());
    int serverPort = server2VM.invoke(() -> new ServerFactory()
        .evictionPolicy(HARegionQueue.HA_EVICTION_POLICY_ENTRY)
        .regionShortcut(RegionShortcut.REPLICATE)
        .create());

    clientVM.invoke(() -> {
      new ClientFactory()
          .cacheListener(new ClientListener())
          .serverPorts(serverPort)
          .create();

      Region<String, DeltaTestImpl> region = getClientCache().getRegion(regionName);
      region.registerInterest("ALL_KEYS");
    });

    server1VM.invoke(() -> {
      Region<String, Object> region = getCache().getRegion(regionName);
      region.create(DELTA_KEY, DELTA_VALUES[0]);
      for (int i = 1; i < EVENTS_SIZE; i++) {
        region.put(DELTA_KEY, DELTA_VALUES[i]);
      }
      region.put(LAST_KEY, "");
    });

    server2VM.invoke(() -> {
      Region<String, Object> region =
          getCache().getRegion(generateNameForClientMsgsRegion(serverPort));

      EvictionController evictionController = ((DiskRecoveryStore) region)
          .getRegionMap()
          .getEvictionController();

      await().untilAsserted(() -> {
        assertThat(evictionController.getCounters().getEvictions()).isGreaterThan(0);
      });

      LATCH.get().countDown();
    });

    clientVM.invoke(() -> {
      await().until(LAST_KEY_RECEIVED::get);

      long toDeltasOnServer1 = server1VM.invoke(DeltaTestImpl::getToDeltaInvokations);
      long fromDeltasOnServer2 = server2VM.invoke(DeltaTestImpl::getFromDeltaInvokations);
      long toDeltasOnServer2 = server2VM.invoke(DeltaTestImpl::getToDeltaInvokations);
      long fromDeltasOnClient = DeltaTestImpl.getFromDeltaInvokations();

      assertThat(toDeltasOnServer1).isEqualTo(EVENTS_SIZE - 1);
      assertThat(fromDeltasOnServer2).isEqualTo(EVENTS_SIZE - 1);
      assertThat(toDeltasOnServer2).isZero();
      assertThat(fromDeltasOnClient).isEqualTo(EVENTS_SIZE - 1);
    });
  }
Tests that an update on a server with full Delta object causes distribution of the full Delta
instance, and not its delta bits, to other peers, even if that instance's
{@code hasDelta()} returns true.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/DeltaPropagationDUnitTest.java