[to #763] Pick bug fixes of API v2 codec to master (#769)

* [cherry-pick] ignore invalid region while decoding EpochNotMatch (#765)

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>
Signed-off-by: Ping Yu <yuping@pingcap.com>

* fix region out of keyspace (#766)

* fix region decoding

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>

* add test for decodeRegion

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>

* add more tests for decodeRegion

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>

---------

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>
Signed-off-by: Ping Yu <yuping@pingcap.com>

* [to #763] Complement v2 codec unit test cases (#768)

* complement v2 codec unit test cases.

Signed-off-by: Ping Yu <yuping@pingcap.com>

* fix fmt

Signed-off-by: Ping Yu <yuping@pingcap.com>

---------

Signed-off-by: Ping Yu <yuping@pingcap.com>

---------

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>
Signed-off-by: Ping Yu <yuping@pingcap.com>
Co-authored-by: iosmanthus <dengliming@pingcap.com>

src/main/java/org/tikv/common/apiversion/CodecUtils.java

@@ -23,7 +23,7 @@ import org.tikv.common.codec.CodecDataInput;
 import org.tikv.common.codec.CodecDataOutput;
 
 // TODO(iosmanthus): use ByteString.wrap to avoid once more copying.
-class CodecUtils {
+public class CodecUtils {
   public static ByteString encode(ByteString key) {
     CodecDataOutput cdo = new CodecDataOutput();
     BytesCodec.writeBytes(cdo, key.toByteArray());

src/main/java/org/tikv/common/apiversion/RequestKeyV2Codec.java

@@ -81,22 +81,21 @@ public class RequestKeyV2Codec implements RequestKeyCodec {
 
     if (!start.isEmpty()) {
       start = CodecUtils.decode(start);
-      if (ByteString.unsignedLexicographicalComparator().compare(start, keyPrefix) < 0) {
-        start = ByteString.EMPTY;
-      } else {
-        start = decodeKey(start);
-      }
     }
 
     if (!end.isEmpty()) {
       end = CodecUtils.decode(end);
-      if (ByteString.unsignedLexicographicalComparator().compare(end, infiniteEndKey) >= 0) {
-        end = ByteString.EMPTY;
-      } else {
-        end = decodeKey(end);
-      }
     }
 
+    if (ByteString.unsignedLexicographicalComparator().compare(start, infiniteEndKey) >= 0
+        || (!end.isEmpty()
+            && ByteString.unsignedLexicographicalComparator().compare(end, keyPrefix) <= 0)) {
+      throw new IllegalArgumentException("region out of keyspace" + region.toString());
+    }
+
+    start = start.startsWith(keyPrefix) ? start.substring(keyPrefix.size()) : ByteString.EMPTY;
+    end = end.startsWith(keyPrefix) ? end.substring(keyPrefix.size()) : ByteString.EMPTY;
+
     return builder.setStartKey(start).setEndKey(end).build();
   }
 }

src/main/java/org/tikv/common/operation/RegionErrorHandler.java

@@ -229,7 +229,16 @@ public class RegionErrorHandler<RespT> implements ErrorHandler<RespT> {
     // If the region epoch is not ahead of TiKV's, replace region meta in region cache.
     for (Metapb.Region meta : currentRegions) {
       // The region needs to be decoded to plain format.
-      meta = regionManager.getPDClient().getCodec().decodeRegion(meta);
+      try {
+        meta = regionManager.getPDClient().getCodec().decodeRegion(meta);
+      } catch (Exception e) {
+        logger.warn("ignore invalid region: " + meta.toString());
+        // if the region is invalid, ignore it since the following situation might appear.
+        // Assuming a region with range [r000, z), then it splits into:
+        // [r000, x) [x, z), the right region is invalid for keyspace `r000`.
+        // We should only care about the valid region.
+        continue;
+      }
       TiRegion region = regionManager.createRegion(meta, backOffer);
       newRegions.add(region);
       if (recv.getRegion().getVerID() == region.getVerID()) {

src/test/java/org/tikv/common/apiversion/RequestKeyCodecTest.java

@@ -17,8 +17,7 @@
 
 package org.tikv.common.apiversion;
 
-import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.*;
-import static org.junit.Assert.assertEquals;
 
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
@@ -177,5 +176,85 @@ public class RequestKeyCodecTest {
     decoded = v2.decodeRegion(region);
     assertEquals(start, decoded.getStartKey());
     assertEquals(ByteString.EMPTY, decoded.getEndKey());
+
+    // test region out of keyspace
+    {
+      ByteString m_123 = CodecUtils.encode(ByteString.copyFromUtf8("m_123"));
+      ByteString m_124 = CodecUtils.encode(ByteString.copyFromUtf8("m_124"));
+      ByteString infiniteEndKey_0 =
+          CodecUtils.encode(v2.infiniteEndKey.concat(ByteString.copyFrom(new byte[] {0})));
+      ByteString t_123 = CodecUtils.encode(ByteString.copyFromUtf8("t_123"));
+      ByteString y_123 = CodecUtils.encode(ByteString.copyFromUtf8("y_123"));
+
+      ByteString[][] outOfKeyspaceCases = {
+        {ByteString.EMPTY, CodecUtils.encode(v2.keyPrefix)}, // ["", "r000"/"x000")
+        {ByteString.EMPTY, m_123},
+        {m_123, m_124},
+        {m_124, CodecUtils.encode(v2.keyPrefix)},
+        {CodecUtils.encode(v2.infiniteEndKey), ByteString.EMPTY}, // ["r001"/"x001", "")
+        {CodecUtils.encode(v2.infiniteEndKey), infiniteEndKey_0},
+        {infiniteEndKey_0, t_123},
+        {y_123, ByteString.EMPTY}, // "y_123" is bigger than "infiniteEndKey" for both raw & txn.
+      };
+
+      for (ByteString[] testCase : outOfKeyspaceCases) {
+        region = Region.newBuilder().setStartKey(testCase[0]).setEndKey(testCase[1]).build();
+        try {
+          decoded = v2.decodeRegion(region);
+          fail(String.format("[%s,%s): %s", testCase[0], testCase[1], decoded.toString()));
+        } catch (Exception ignored) {
+        }
+      }
+    }
+
+    // case: regionStartKey == "" < keyPrefix < regionEndKey < infiniteEndKey
+    region =
+        Region.newBuilder()
+            .setStartKey(ByteString.EMPTY)
+            .setEndKey(CodecUtils.encode(v2.keyPrefix.concat(ByteString.copyFromUtf8("0"))))
+            .build();
+    decoded = v2.decodeRegion(region);
+    assertTrue(decoded.getStartKey().isEmpty());
+    assertEquals(ByteString.copyFromUtf8("0"), decoded.getEndKey());
+
+    // case: "" < regionStartKey < keyPrefix < regionEndKey < infiniteEndKey < ""
+    region =
+        Region.newBuilder()
+            .setStartKey(CodecUtils.encode(ByteString.copyFromUtf8("m_123")))
+            .setEndKey(CodecUtils.encode(v2.keyPrefix.concat(ByteString.copyFromUtf8("0"))))
+            .build();
+    decoded = v2.decodeRegion(region);
+    assertEquals(ByteString.EMPTY, decoded.getStartKey());
+    assertEquals(ByteString.copyFromUtf8("0"), decoded.getEndKey());
+
+    // case: "" < regionStartKey < keyPrefix < infiniteEndKey < regionEndKey < ""
+    region =
+        Region.newBuilder()
+            .setStartKey(CodecUtils.encode(ByteString.copyFromUtf8("m_123")))
+            .setEndKey(CodecUtils.encode(v2.infiniteEndKey.concat(ByteString.copyFromUtf8("0"))))
+            .build();
+    decoded = v2.decodeRegion(region);
+    assertEquals(ByteString.EMPTY, decoded.getStartKey());
+    assertEquals(ByteString.EMPTY, decoded.getEndKey());
+
+    // case: keyPrefix < regionStartKey <  infiniteEndKey < regionEndKey < ""
+    region =
+        Region.newBuilder()
+            .setStartKey(CodecUtils.encode(v2.keyPrefix.concat(ByteString.copyFromUtf8("0"))))
+            .setEndKey(CodecUtils.encode(v2.infiniteEndKey.concat(ByteString.copyFromUtf8("0"))))
+            .build();
+    decoded = v2.decodeRegion(region);
+    assertEquals(ByteString.copyFromUtf8("0"), decoded.getStartKey());
+    assertTrue(decoded.getEndKey().isEmpty());
+
+    // case: keyPrefix < regionStartKey <  infiniteEndKey < regionEndKey == ""
+    region =
+        Region.newBuilder()
+            .setStartKey(CodecUtils.encode(v2.keyPrefix.concat(ByteString.copyFromUtf8("0"))))
+            .setEndKey(ByteString.EMPTY)
+            .build();
+    decoded = v2.decodeRegion(region);
+    assertEquals(ByteString.copyFromUtf8("0"), decoded.getStartKey());
+    assertTrue(decoded.getEndKey().isEmpty());
   }
 }