  @Test
  public void testFallback() throws TableNotExistException, NetworkException {

    final long origOpId = 0;

    doAnswer(invocation -> {
      final Object[] arguments = invocation.getArguments();
      final String origId = (String) arguments[0];
      final String destId = (String) arguments[1];
      final long opId = (long) arguments[2];

      assertEquals("The origin ID should be kept same", origId, ORIG_EXECUTOR_ID);
      assertEquals("The request should be redirected to the driver", DRIVER_ID, destId);
      assertEquals("The op ID should be kept same", origOpId, opId);
      return null;
    }).when(mockMsgSender).sendTableAccessReqMsg(anyString(), anyString(), anyLong(), anyString(),
        any(OpType.class), anyBoolean(), any(DataKey.class), anyObject());

    final TableComponents<String, Integer, Integer> tableComponents = tables.getTableComponents(TABLE_ID);
    final KVUSerializer<String, Integer, Integer> kvuSerializer = tableComponents.getSerializer();
    final StreamingCodec<String> keyCodec = kvuSerializer.getKeyCodec();
    final StreamingCodec<Integer> valueCodec = kvuSerializer.getValueCodec();

    // Assume that the table has been unassociated
    tables.remove(TABLE_ID);

    // Any operation will give the same results; GET is used here for simplicity.
    final String key = "key";
    final Integer value = 1;
    final Pair<DataKey, DataValue> dataPair = getDataPair(key, value, keyCodec, valueCodec);

    final TableAccessReqMsg getMsg = TableAccessReqMsg.newBuilder()
        .setOrigId(ORIG_EXECUTOR_ID)
        .setOpType(OpType.GET)
        .setTableId(TABLE_ID)
        .setDataKey(dataPair.getKey())
        .setDataValue(null)
        .build();

    remoteAccessOpHandler.onTableAccessReqMsg(origOpId, getMsg);
  }