@Test
  public void testPutAllReturnsExceptions() {
    // set <true, false> means <PR=true, notifyBySubscription=false> to test local-invalidates
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());
    server2.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1)
        .create());

    // server1 add cacheWriter
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // let the server to trigger exception after created 15 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                if (creates >= 15) {
                  throw new CacheWriterException("Expected by test");
                }
              })));

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown).isInstanceOf(CacheWriterException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown).isInstanceOf(CacheWriterException.class);
    });

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      String message =
          String.format("Region %s putAll at server applied partial keys due to exception.",
              region.getFullPath());

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));

      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasMessageContaining(message)
          .hasCauseInstanceOf(CacheWriterException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasCauseInstanceOf(CacheWriterException.class);
    });

    // let server1 to throw CancelException
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // let the server to trigger exception after created 15 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                throw new OplogCancelledException("Expected by test");
              })));

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown).isInstanceOf(OplogCancelledException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown).isInstanceOf(OplogCancelledException.class);
    });

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown)
          .isInstanceOf(ServerConnectivityException.class)
          .hasCauseInstanceOf(OplogCancelledException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(ServerConnectivityException.class)
          .hasCauseInstanceOf(OplogCancelledException.class);
    });

    // let server1 to throw LowMemoryException
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // let the server to trigger exception after created 15 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                throw new LowMemoryException("Testing", emptySet());
              })));

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown).isInstanceOf(LowMemoryException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown).isInstanceOf(LowMemoryException.class);
    });

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .getCause()
          .isInstanceOf(LowMemoryException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .getCause()
          .isInstanceOf(LowMemoryException.class);
    });

    // let server1 to throw TimeoutException
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // let the server to trigger exception after created 15 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                throw new TimeoutException("Testing");
              })));

      Map<String, TickerData> map = new LinkedHashMap<>();
      for (int i = 0; i < ONE_HUNDRED; i++) {
        map.put(keyPrefix + i, new TickerData(i));
      }

      Throwable thrown = catchThrowable(() -> region.putAll(map));
      assertThat(thrown)
          .isInstanceOf(TimeoutException.class)
          .hasNoCause();

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(TimeoutException.class)
          .hasNoCause();

      thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown)
          .isInstanceOf(TimeoutException.class)
          .hasNoCause();

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(TimeoutException.class)
          .hasNoCause();
    });

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      Map<String, TickerData> map = new LinkedHashMap<>();
      for (int i = 0; i < ONE_HUNDRED; i++) {
        map.put(keyPrefix + i, new TickerData(i));
      }

      Throwable thrown = catchThrowable(() -> region.putAll(map));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasCauseInstanceOf(TimeoutException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasCauseInstanceOf(TimeoutException.class);

      thrown = catchThrowable(() -> doPutAll(region, "key-", ONE_HUNDRED));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasCauseInstanceOf(TimeoutException.class);

      thrown = catchThrowable(() -> region.put("dummyKey", new TickerData(0)));
      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasCauseInstanceOf(TimeoutException.class);
    });
  }
Verify all the possible exceptions a putAll/put/removeAll/destroy could return
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java