diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java
index 3592003..605547b 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java
@@ -39,6 +39,7 @@
 import static org.mockito.Mockito.spy;
 
 import java.io.File;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -50,9 +51,14 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -70,13 +76,16 @@
 import org.apache.geode.cache.CacheLoader;
 import org.apache.geode.cache.CacheLoaderException;
 import org.apache.geode.cache.CacheWriterException;
+import org.apache.geode.cache.CommitConflictException;
 import org.apache.geode.cache.DiskStoreFactory;
 import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.EntryNotFoundException;
 import org.apache.geode.cache.EvictionAttributes;
 import org.apache.geode.cache.LoaderHelper;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.TransactionDataRebalancedException;
 import org.apache.geode.cache.asyncqueue.AsyncEvent;
 import org.apache.geode.cache.asyncqueue.AsyncEventListener;
 import org.apache.geode.cache.control.RebalanceOperation;
@@ -98,10 +107,14 @@
 import org.apache.geode.internal.cache.PRHARedundancyProvider;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.PartitionedRegionDataStore;
+import org.apache.geode.internal.cache.RegionEntry;
 import org.apache.geode.internal.cache.SignalBounceOnRequestImageMessageObserver;
+import org.apache.geode.internal.cache.TXManagerImpl;
 import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserverAdapter;
 import org.apache.geode.internal.cache.partitioned.BucketCountLoadProbe;
 import org.apache.geode.internal.cache.partitioned.LoadProbe;
+import org.apache.geode.internal.cache.versions.VersionStamp;
+import org.apache.geode.internal.cache.versions.VersionTag;
 import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.SerializableRunnable;
@@ -125,6 +138,12 @@
 
   private static final long TIMEOUT_SECONDS = GeodeAwaitility.getTimeout().getSeconds();
 
+  private VM vm0 = getVM(0);
+  private VM vm1 = getVM(1);
+  private VM vm2 = getVM(2);
+  private VM vm3 = getVM(3);
+  private boolean toSetBucketNumber = false;
+
   @Rule
   public DistributedRestoreSystemProperties restoreSystemProperties =
       new DistributedRestoreSystemProperties();
@@ -144,9 +163,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testRecoverRedundancy(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     // Create the region in only 1 VM
     vm0.invoke(() -> createPartitionedRegion("region1", 1));
 
@@ -215,9 +231,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testEnforceIP(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     for (VM vm : toArray(vm0, vm1)) {
       vm.invoke(() -> {
         Properties props = new Properties();
@@ -286,10 +299,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testEnforceZone(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     vm0.invoke(() -> setRedundancyZone("A"));
     vm1.invoke(() -> setRedundancyZone("A"));
 
@@ -371,10 +380,6 @@
 
   @Test
   public void testEnforceZoneWithMultipleRegions() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     vm0.invoke(() -> setRedundancyZone("A"));
     vm1.invoke(() -> setRedundancyZone("A"));
 
@@ -463,10 +468,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testRecoverRedundancyBalancing(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     DistributedMember member1 = vm0.invoke(() -> {
       createPartitionedRegion("region1", 1, 200);
       return getCache().getDistributedSystem().getDistributedMember();
@@ -534,10 +535,6 @@
 
   @Test
   public void testRecoverRedundancyBalancingIfCreateBucketFails() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     DistributedMember member1 = vm0.invoke(() -> {
       createPartitionedRegion("region1", 1, 100);
       return getCache().getDistributedSystem().getDistributedMember();
@@ -644,10 +641,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testRecoverRedundancyColocatedRegions(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     DistributedMember member1 = vm0.invoke(() -> {
       createPartitionedRegion("region1", 1, 200);
       return getCache().getDistributedSystem().getDistributedMember();
@@ -751,10 +744,6 @@
           () -> System.setProperty(GeodeGlossary.GEMFIRE_PREFIX + "LOG_REBALANCE", "true"));
     }
 
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     DistributedMember member1 = vm0.invoke(() -> {
       createPRRegionWithAsyncQueue(200);
       return getCache().getDistributedSystem().getDistributedMember();
@@ -857,9 +846,6 @@
 
   @Test
   public void testCancelOperation() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     // Create the region in only 1 VM
     vm0.invoke(() -> createPartitionedRegion("region1", 1));
 
@@ -935,10 +921,6 @@
    */
   @Test
   public void testMembershipChange() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     // Create the region in only 1 VM
     vm0.invoke(() -> createPartitionedRegion("region1", 0));
 
@@ -1018,9 +1000,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testMoveBucketsNoRedundancy(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     // Create the region in only 1 VM
     vm0.invoke(() -> createPartitionedRegion("region1", new NullReturningLoader()));
 
@@ -1116,9 +1095,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testFilterRegions(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     Set<String> included = new HashSet<>();
     included.add("region0");
     included.add("region1");
@@ -1238,10 +1214,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testMoveBucketsWithRedundancy(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     // Create the region in two VMs
     vm0.invoke(() -> createPartitionedRegion("region1", 1));
     vm1.invoke(() -> createPartitionedRegion("region1", 1));
@@ -1332,11 +1304,6 @@
    */
   @Test
   public void testMoveBucketsOverflowToDisk() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-    VM vm3 = getVM(3);
-
     // Create the region in two VMs
     vm0.invoke(
         () -> createPartitionedRegion("region1", createLRUEntryAttributes(1, OVERFLOW_TO_DISK)));
@@ -1501,10 +1468,6 @@
    */
   @Test
   public void testMoveBucketsNestedPR() {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     // Create the region in two VMs
     for (VM vm : toArray(vm0, vm1)) {
       vm.invoke(() -> {
@@ -1597,10 +1560,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testMoveBucketsColocatedRegions(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-
     vm0.invoke(() -> createPartitionedRegion("region1", 1, 200));
     vm0.invoke(() -> createPartitionedRegion("region2", 200, "region1"));
     vm1.invoke(() -> createPartitionedRegion("region1", 1, 200));
@@ -1707,8 +1666,6 @@
   @Parameters({"PUT", "INVALIDATE", "DESTROY", "CACHE_LOADER"})
   @TestCaseName("{method}(operation={0})")
   public void runTestWaitForOperation(OperationEnum operation) throws Exception {
-    VM vm1 = getVM(1);
-
     // Create a region in this VM with a cache writer
     // and cache loader
     createPartitionedRegion("region1", 1, 100, (CacheLoader<Number, String>) helper1 -> "anobject");
@@ -1780,11 +1737,6 @@
   @TestCaseName("{method}(simulate={0}, userAccessor={1})")
   public void testRecoverRedundancyWithOfflinePersistence(boolean simulate, boolean useAccessor)
       throws Exception {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-    VM vm2 = getVM(2);
-    VM vm3 = getVM(3);
-
     int redundantCopies = 1;
 
     // Create the region in only 2 VMs
@@ -1967,9 +1919,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testMoveBucketsWithUnrecoveredValues(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     // Create the region in only 1 VM
     vm0.invoke(() -> createPersistentPartitionedRegion("region1", "store", getDiskDirs(),
         new NullReturningLoader<>()));
@@ -2102,9 +2051,6 @@
   @Parameters({"true", "false"})
   @TestCaseName("{method}(simulate={0})")
   public void testBalanceBucketsByCount(boolean simulate) {
-    VM vm0 = getVM(0);
-    VM vm1 = getVM(1);
-
     // Cache is closed so loadProbeToRestore does not need to be restored
     vm0.invoke(
         () -> {
@@ -2356,6 +2302,9 @@
     partitionAttributesFactory.setRedundantCopies(redundantCopies);
     partitionAttributesFactory.setRecoveryDelay(-1);
     partitionAttributesFactory.setStartupRecoveryDelay(-1);
+    if (toSetBucketNumber) {
+      partitionAttributesFactory.setTotalNumBuckets(totalNumberOfBuckets);
+    }
 
     RegionFactory regionFactory = getCache().createRegionFactory(PARTITION);
     regionFactory.setPartitionAttributes(partitionAttributesFactory.create());
@@ -2834,4 +2783,180 @@
       closed = true;
     }
   }
+
+  private String regionName = "region";
+  private int numOfEntry = 2500;
+  private int totalNumberOfBuckets = 31;
+  private final ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
+
+  @Test
+  public void correctVersionGeneratedForConcurrentOperationsInTxWithRebalance() throws Exception {
+    toSetBucketNumber = true;
+    vm0.invoke(() -> createPartitionedRegion(regionName, 0));
+    vm0.invoke(() -> doPut("A"));
+    vm0.invoke(this::doDestroy);
+    vm0.invoke(() -> doPut("B"));
+
+    vm1.invoke(() -> createPartitionedRegion(regionName, 0));
+    doConcurrentOpsAndRebalance("C");
+    validateVersionsInVms(vm0, vm1);
+
+    vm2.invoke(() -> createPartitionedRegion(regionName, 0));
+    doConcurrentOpsAndRebalance("D");
+    validateVersionsInVms(vm0, vm1, vm2);
+
+    vm3.invoke(() -> createPartitionedRegion(regionName, 0));
+    doConcurrentOpsAndRebalance("E");
+    validateVersionsInVms(vm0, vm1, vm2, vm3);
+  }
+
+  private void doConcurrentOpsAndRebalance(String s) throws Exception {
+    AsyncInvocation async0 = vm0.invokeAsync(this::doConcurrentDestroyInTx);
+    AsyncInvocation async1 = vm1.invokeAsync(() -> doConcurrentPutInTx(s));
+    vm0.invoke(() -> doRebalance());
+    async0.await();
+    async1.await();
+  }
+
+  private void doRebalance() throws TimeoutException, InterruptedException {
+    InternalResourceManager manager = getCache().getInternalResourceManager();
+    doRebalance(false, manager);
+  }
+
+  private void doConcurrentPutInTx(String s) throws Exception {
+    for (int i = 0; i < totalNumberOfBuckets; i++) {
+      queue.add(i);
+    }
+
+    ExecutorService pool = Executors.newCachedThreadPool();
+    Collection<Callable<Object>> tasks = new ArrayList<>();
+    Callable<Object> task = () -> {
+      doPutOpInTx(s);
+      return null;
+    };
+    for (int i = 0; i < totalNumberOfBuckets; i++) {
+      tasks.add(task);
+    }
+
+    List<Future<Object>> futures = pool.invokeAll(tasks);
+    for (Future future : futures) {
+      future.get();
+    }
+  }
+
+  private void doConcurrentDestroyInTx() throws Exception {
+    for (int i = 0; i < totalNumberOfBuckets; i++) {
+      queue.add(i);
+    }
+
+    ExecutorService pool = Executors.newCachedThreadPool();
+    Collection<Callable<Object>> tasks = new ArrayList<>();
+    Callable<Object> task = () -> {
+      doDestroyOpInTx();
+      return null;
+    };
+    for (int i = 0; i < totalNumberOfBuckets; i++) {
+      tasks.add(task);
+    }
+
+    List<Future<Object>> futures = pool.invokeAll(tasks);
+    for (Future future : futures) {
+      future.get();
+    }
+  }
+
+  private void doPutOpInTx(String s) {
+    int bucket;
+    if (!queue.isEmpty()) {
+      bucket = queue.poll();
+      Region<Number, String> region = getCache().getRegion(regionName);
+      for (int i = 0; i < numOfEntry; i++) {
+        if (i % totalNumberOfBuckets == bucket) {
+          doTXPut(region, i, s);
+        }
+      }
+    }
+  }
+
+  private void doDestroyOpInTx() {
+    int bucket;
+    if (!queue.isEmpty()) {
+      bucket = queue.poll();
+      Region<Number, String> region = getCache().getRegion(regionName);
+      for (int i = 0; i < numOfEntry; i++) {
+        if (i % totalNumberOfBuckets == bucket) {
+          doTxDestroy(region, i);
+        }
+      }
+    }
+  }
+
+  private void doPut(String s) {
+    Region<Number, String> region = getCache().getRegion(regionName);
+    for (int i = 0; i < numOfEntry; i++) {
+      region.put(i, s);
+    }
+  }
+
+  private void doTXPut(Region<Number, String> region, int i, String s) {
+    TXManagerImpl manager = getCache().getTxManager();
+    manager.begin();
+    try {
+      region.put(i, s);
+      manager.commit();
+    } catch (TransactionDataRebalancedException e) {
+      if (manager.getTransactionId() != null) {
+        manager.rollback();
+      }
+    } catch (CommitConflictException ignore) {
+    }
+  }
+
+  private void doTxDestroy(Region<Number, String> region, int i) {
+    TXManagerImpl manager = getCache().getTxManager();
+    manager.begin();
+    try {
+      region.remove(i);
+      manager.commit();
+    } catch (TransactionDataRebalancedException e) {
+      if (manager.getTransactionId() != null) {
+        manager.rollback();
+      }
+    } catch (CommitConflictException ignore) {
+    }
+  }
+
+  private void doDestroy() {
+    Region<Number, String> region = getCache().getRegion(regionName);
+    for (int i = 0; i < numOfEntry; i++) {
+      try {
+        region.destroy(i);
+      } catch (EntryNotFoundException ignore) {
+      }
+    }
+  }
+
+  private void validateVersionsInVms(VM... vms) {
+    for (VM vm : vms) {
+      vm.invoke(this::validateEntryVersions);
+    }
+  }
+
+  private void validateEntryVersions() {
+    PartitionedRegion region = (PartitionedRegion) getCache().getRegion(regionName);
+    for (int i = 0; i < numOfEntry; i++) {
+      BucketRegion bucketRegion = region.getDataStore().getLocalBucketByKey(i);
+      if (bucketRegion != null) {
+        RegionEntry entry = bucketRegion.getRegionMap().getEntry(i);
+        if (entry != null) {
+          VersionStamp stamp = entry.getVersionStamp();
+          VersionTag tag = stamp.asVersionTag();
+          if (tag.getEntryVersion() < 3) {
+            logger.info("tag for key {} is " + tag, i);
+          }
+          assertThat(tag.getEntryVersion()).isGreaterThan(2);
+        }
+      }
+    }
+  }
 }
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 61328f3..017329c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -6836,7 +6836,7 @@
     final boolean needRIDestroyToken = inRI && riCnt > 0;
 
     try {
-      final boolean inTokenMode = needTokensForGII || needRIDestroyToken;
+      final boolean inTokenMode = isInTokenModeNeeded(needTokensForGII, needRIDestroyToken);
       entries.txApplyDestroy(key, rmtOrigin, event, inTokenMode, needRIDestroyToken, op,
           eventId, aCallbackArgument, pendingCallbacks, filterRoutingInfo, bridgeContext,
           isOriginRemote, txEntryState, versionTag, tailKey);
@@ -6847,6 +6847,10 @@
     }
   }
 
+  boolean isInTokenModeNeeded(boolean needTokensForGII, boolean needRIDestroyToken) {
+    return !getConcurrencyChecksEnabled() && (needTokensForGII || needRIDestroyToken);
+  }
+
   /**
    * Called by lower levels, while still holding the write sync lock, and the low level has
    * completed its part of the basic destroy
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java
index dd9cce1..60e8454 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java
@@ -277,4 +277,45 @@
 
     assertThat(region.isGenerateLocalFilterRoutingNeeded(event)).isFalse();
   }
+
+  @Test
+  public void isInTokenModeNeededReturnsFalseIfConcurrencyChecksEnabled() {
+    LocalRegion region =
+        spy(new LocalRegion("region", regionAttributes, null, cache, internalRegionArguments,
+            internalDataView, regionMapConstructor, serverRegionProxyConstructor, entryEventFactory,
+            poolFinder, regionPerfStatsFactory, disabledClock()));
+    doReturn(true).when(region).getConcurrencyChecksEnabled();
+
+    assertThat(region.isInTokenModeNeeded(true, true)).isFalse();
+  }
+
+  @Test
+  public void isInTokenModeNeededReturnsFalseIfBothNeedTokensForGIIAndNeedRIDestroyTokenAreFalse() {
+    LocalRegion region =
+        spy(new LocalRegion("region", regionAttributes, null, cache, internalRegionArguments,
+            internalDataView, regionMapConstructor, serverRegionProxyConstructor, entryEventFactory,
+            poolFinder, regionPerfStatsFactory, disabledClock()));
+
+    assertThat(region.isInTokenModeNeeded(false, false)).isFalse();
+  }
+
+  @Test
+  public void isInTokenModeNeededReturnsTrueIfConcurrencyChecksNotEnabledAndNeedTokensForGII() {
+    LocalRegion region =
+        spy(new LocalRegion("region", regionAttributes, null, cache, internalRegionArguments,
+            internalDataView, regionMapConstructor, serverRegionProxyConstructor, entryEventFactory,
+            poolFinder, regionPerfStatsFactory, disabledClock()));
+
+    assertThat(region.isInTokenModeNeeded(true, false)).isTrue();
+  }
+
+  @Test
+  public void isInTokenModeNeededReturnsTrueIfConcurrencyChecksNotEnabledAndNeedRIDestroyToken() {
+    LocalRegion region =
+        spy(new LocalRegion("region", regionAttributes, null, cache, internalRegionArguments,
+            internalDataView, regionMapConstructor, serverRegionProxyConstructor, entryEventFactory,
+            poolFinder, regionPerfStatsFactory, disabledClock()));
+
+    assertThat(region.isInTokenModeNeeded(false, true)).isTrue();
+  }
 }