content: /*
   * Copyright 1999-2023 Alibaba Group Holding Ltd.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.alibaba.nacos.core.persistence;
  
  import com.alibaba.nacos.api.exception.NacosException;
  import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
  import com.alibaba.nacos.common.JustForTest;
  import com.alibaba.nacos.common.model.RestResult;
  import com.alibaba.nacos.common.model.RestResultUtils;
  import com.alibaba.nacos.common.notify.Event;
  import com.alibaba.nacos.common.notify.NotifyCenter;
  import com.alibaba.nacos.common.notify.listener.Subscriber;
  import com.alibaba.nacos.common.utils.ExceptionUtil;
  import com.alibaba.nacos.common.utils.LoggerUtils;
  import com.alibaba.nacos.common.utils.MD5Utils;
  import com.alibaba.nacos.common.utils.Preconditions;
  import com.alibaba.nacos.common.utils.StringUtils;
  import com.alibaba.nacos.consistency.SerializeFactory;
  import com.alibaba.nacos.consistency.Serializer;
  import com.alibaba.nacos.consistency.cp.CPProtocol;
  import com.alibaba.nacos.consistency.cp.RequestProcessor4CP;
  import com.alibaba.nacos.consistency.entity.ReadRequest;
  import com.alibaba.nacos.consistency.entity.Response;
  import com.alibaba.nacos.consistency.entity.WriteRequest;
  import com.alibaba.nacos.consistency.exception.ConsistencyException;
  import com.alibaba.nacos.consistency.snapshot.SnapshotOperation;
  import com.alibaba.nacos.core.cluster.ServerMemberManager;
  import com.alibaba.nacos.core.distributed.ProtocolManager;
  import com.alibaba.nacos.core.utils.ClassUtils;
  import com.alibaba.nacos.persistence.configuration.condition.ConditionDistributedEmbedStorage;
  import com.alibaba.nacos.persistence.constants.PersistenceConstant;
  import com.alibaba.nacos.persistence.datasource.DynamicDataSource;
  import com.alibaba.nacos.persistence.datasource.LocalDataSourceServiceImpl;
  import com.alibaba.nacos.persistence.exception.NJdbcException;
  import com.alibaba.nacos.persistence.model.event.DerbyLoadEvent;
  import com.alibaba.nacos.persistence.model.event.RaftDbErrorEvent;
  import com.alibaba.nacos.persistence.repository.RowMapperManager;
  import com.alibaba.nacos.persistence.repository.embedded.EmbeddedStorageContextHolder;
  import com.alibaba.nacos.persistence.repository.embedded.hook.EmbeddedApplyHook;
  import com.alibaba.nacos.persistence.repository.embedded.hook.EmbeddedApplyHookHolder;
  import com.alibaba.nacos.persistence.repository.embedded.operate.BaseDatabaseOperate;
  import com.alibaba.nacos.persistence.repository.embedded.sql.ModifyRequest;
  import com.alibaba.nacos.persistence.repository.embedded.sql.QueryType;
  import com.alibaba.nacos.persistence.repository.embedded.sql.SelectRequest;
  import com.alibaba.nacos.persistence.repository.embedded.sql.limiter.SqlLimiter;
  import com.alibaba.nacos.persistence.repository.embedded.sql.limiter.SqlTypeLimiter;
  import com.alibaba.nacos.persistence.utils.PersistenceExecutor;
  import com.alibaba.nacos.sys.utils.DiskUtils;
  import com.google.protobuf.ByteString;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.context.annotation.Conditional;
  import org.springframework.dao.DataAccessException;
  import org.springframework.dao.DataIntegrityViolationException;
  import org.springframework.jdbc.BadSqlGrammarException;
  import org.springframework.jdbc.core.JdbcTemplate;
  import org.springframework.jdbc.core.RowMapper;
  import org.springframework.stereotype.Component;
  import org.springframework.transaction.support.TransactionTemplate;
  
  import java.io.File;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import java.util.function.BiConsumer;
  import java.util.stream.Collectors;
  
  /**
   * Distributed Database Operate.
   *
   * <pre>
   *                   ┌────────────────────┐
   *               ┌──▶│   PersistService   │
   *               │   └────────────────────┘ ┌─────────────────┐
   *               │              │           │                 │
   *               │              │           │                 ▼
   *               │              │           │    ┌────────────────────────┐
   *               │              │           │    │  acquireSnowFlowerId   │
   *               │              │           │    └────────────────────────┘
   *               │              │           │                 │
   *               │              │           │                 │
   *               │              │           │                 ▼
   *               │              │           │      ┌────────────────────┐          save sql
   *               │              ▼           │      │     saveConfig     │──────────context─────────────┐
   *               │     ┌────────────────┐   │      └────────────────────┘                              │
   *               │     │ publishConfig  │───┘                 │                                        │
   *               │     └────────────────┘                     │                                        │
   *               │                                            ▼                                        ▼
   *               │                               ┌─────────────────────────┐    save sql    ┌────────────────────┐
   *               │                               │ saveConfigTagRelations  │────context────▶│  SqlContextUtils   │◀─┐
   *        publish config                         └─────────────────────────┘                └────────────────────┘  │
   *               │                                            │                                        ▲            │
   *               │                                            │                                        │            │
   *               │                                            ▼                                        │            │
   *               │                                ┌───────────────────────┐         save sql           │            │
   *            ┌────┐                              │   saveConfigHistory   │─────────context────────────┘            │
   *            │user│                              └───────────────────────┘                                         │
   *            └────┘                                                                                                │
   *               ▲                                                                                                  │
   *               │                                           ┌1:getCurrentSqlContexts───────────────────────────────┘
   *               │                                           │
   *               │                                           │
   *               │                                           │
   *               │           ┌───────────────┐    ┌─────────────────────┐
   *               │           │ JdbcTemplate  │◀───│   DatabaseOperate   │───┐
   *       4:execute result    └───────────────┘    └─────────────────────┘   │
   *               │                   │                       ▲              │
   *               │                   │                       │              │
   *               │                   │                  3:onApply         2:submit(List&lt;ModifyRequest&gt;)
   *               │                   │                       │              │
   *               │                   ▼                       │              │
   *               │           ┌──────────────┐                │              │
   *               │           │ Apache Derby │    ┌───────────────────────┐  │
   *               │           └──────────────┘    │     JRaftProtocol     │◀─┘
   *               │                               └───────────────────────┘
   *               │                                           │
   *               │                                           │
   *               └───────────────────────────────────────────┘
   * </pre>
   *
   * <p>TODO depend on Member and Cp protocol strongly, Waiting for addition split.
   *
   * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
   */
  @Conditional(ConditionDistributedEmbedStorage.class)
  @Component
  @SuppressWarnings({"unchecked"})
  public class DistributedDatabaseOperateImpl extends RequestProcessor4CP implements BaseDatabaseOperate {
      
      private static final Logger LOGGER = LoggerFactory.getLogger(DistributedDatabaseOperateImpl.class);
      
      /**
       * The data import operation is dedicated key, which ACTS as an identifier.
       */
      private static final String DATA_IMPORT_KEY = "00--0-data_import-0--00";
      
      private final ServerMemberManager memberManager;
      
      private CPProtocol protocol;
      
      private LocalDataSourceServiceImpl dataSourceService;
      
      private JdbcTemplate jdbcTemplate;
      
      private TransactionTemplate transactionTemplate;
      
      private final Serializer serializer = SerializeFactory.getDefault();
      
      private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
      
      private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
      
      private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
      
      private final SqlLimiter sqlLimiter;
      
      public DistributedDatabaseOperateImpl(ServerMemberManager memberManager, ProtocolManager protocolManager)
              throws Exception {
          this.memberManager = memberManager;
          this.protocol = protocolManager.getCpProtocol();
          init();
          this.sqlLimiter = new SqlTypeLimiter();
      }
      
      protected void init() throws Exception {
          
          this.dataSourceService = (LocalDataSourceServiceImpl) DynamicDataSource.getInstance().getDataSource();
          
          // Because in Raft + Derby mode, ensuring data consistency depends on the Raft's
          // log playback and snapshot recovery capabilities, and the last data must be cleared
          this.dataSourceService.cleanAndReopenDerby();
          
          this.jdbcTemplate = dataSourceService.getJdbcTemplate();
          this.transactionTemplate = dataSourceService.getTransactionTemplate();
          
          // Registers a Derby Raft state machine failure event for node degradation processing
          NotifyCenter.registerToSharePublisher(RaftDbErrorEvent.class);
          // Register the snapshot load event
          NotifyCenter.registerToSharePublisher(DerbyLoadEvent.class);
          
          NotifyCenter.registerSubscriber(new Subscriber<RaftDbErrorEvent>() {
              @Override
              public void onEvent(RaftDbErrorEvent event) {
                  dataSourceService.setHealthStatus("DOWN");
              }
              
              @Override
              public Class<? extends Event> subscribeType() {
                  return RaftDbErrorEvent.class;
              }
          });
          
          this.protocol.addRequestProcessors(Collections.singletonList(this));
          LOGGER.info("use DistributedTransactionServicesImpl");
      }
      
      @JustForTest
      public void mockConsistencyProtocol(CPProtocol protocol) {
          this.protocol = protocol;
      }
      
      @Override
      public <R> R queryOne(String sql, Class<R> cls) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryOne info : sql : {}", sql);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_ONE_NO_MAPPER_NO_ARGS).sql(sql)
                              .className(cls.getCanonicalName()).build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(), cls);
              }
              throw new NJdbcException(response.getErrMsg(), response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public <R> R queryOne(String sql, Object[] args, Class<R> cls) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryOne info : sql : {}, args : {}", sql, args);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_ONE_NO_MAPPER_WITH_ARGS).sql(sql).args(args)
                              .className(cls.getCanonicalName()).build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(), cls);
              }
              throw new NJdbcException(response.getErrMsg(), response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public <R> R queryOne(String sql, Object[] args, RowMapper<R> mapper) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryOne info : sql : {}, args : {}", sql, args);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_ONE_WITH_MAPPER_WITH_ARGS).sql(sql).args(args)
                              .className(mapper.getClass().getCanonicalName()).build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(),
                          ClassUtils.resolveGenericTypeByInterface(mapper.getClass()));
              }
              throw new NJdbcException(response.getErrMsg(), response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public <R> List<R> queryMany(String sql, Object[] args, RowMapper<R> mapper) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryMany info : sql : {}, args : {}", sql, args);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_MANY_WITH_MAPPER_WITH_ARGS).sql(sql).args(args)
                              .className(mapper.getClass().getCanonicalName()).build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(), List.class);
              }
              throw new NJdbcException(response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public <R> List<R> queryMany(String sql, Object[] args, Class<R> rClass) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryMany info : sql : {}, args : {}", sql, args);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_MANY_NO_MAPPER_WITH_ARGS).sql(sql).args(args)
                              .className(rClass.getCanonicalName()).build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(), List.class);
              }
              throw new NJdbcException(response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public List<Map<String, Object>> queryMany(String sql, Object[] args) {
          try {
              LoggerUtils.printIfDebugEnabled(LOGGER, "queryMany info : sql : {}, args : {}", sql, args);
              
              byte[] data = serializer.serialize(
                      SelectRequest.builder().queryType(QueryType.QUERY_MANY_WITH_LIST_WITH_ARGS).sql(sql).args(args)
                              .build());
              
              final boolean blockRead = EmbeddedStorageContextHolder
                      .containsExtendInfo(PersistenceConstant.EXTEND_NEED_READ_UNTIL_HAVE_DATA);
              
              Response response = innerRead(
                      ReadRequest.newBuilder().setGroup(group()).setData(ByteString.copyFrom(data)).build(), blockRead);
              if (response.getSuccess()) {
                  return serializer.deserialize(response.getData().toByteArray(), List.class);
              }
              throw new NJdbcException(response.getErrMsg());
          } catch (Exception e) {
              LOGGER.error("An exception occurred during the query operation : {}", e.toString());
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      /**
       * In some business situations, you need to avoid the timeout issue, so blockRead is used to determine this.
       *
       * @param request   {@link ReadRequest}
       * @param blockRead is async read operation
       * @return {@link Response}
       * @throws Exception Exception
       */
      private Response innerRead(ReadRequest request, boolean blockRead) throws Exception {
          if (blockRead) {
              return (Response) protocol.aGetData(request).join();
          }
          return protocol.getData(request);
      }
      
      @Override
      public CompletableFuture<RestResult<String>> dataImport(File file) {
          return CompletableFuture.supplyAsync(() -> {
              try (DiskUtils.LineIterator iterator = DiskUtils.lineIterator(file)) {
                  int batchSize = 1000;
                  List<String> batchUpdate = new ArrayList<>(batchSize);
                  List<CompletableFuture<Response>> futures = new ArrayList<>();
                  while (iterator.hasNext()) {
                      String sql = iterator.next();
                      if (StringUtils.isNotBlank(sql)) {
                          batchUpdate.add(sql);
                      }
                      boolean submit = batchUpdate.size() == batchSize || !iterator.hasNext();
                      if (submit) {
                          List<ModifyRequest> requests = batchUpdate.stream().map(ModifyRequest::new)
                                  .collect(Collectors.toList());
                          CompletableFuture<Response> future = protocol.writeAsync(
                                  WriteRequest.newBuilder().setGroup(group())
                                          .setData(ByteString.copyFrom(serializer.serialize(requests)))
                                          .putExtendInfo(DATA_IMPORT_KEY, Boolean.TRUE.toString()).build());
                          futures.add(future);
                          batchUpdate.clear();
                      }
                  }
                  CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
                  for (CompletableFuture<Response> future : futures) {
                      Response response = future.get();
                      if (!response.getSuccess()) {
                          return RestResultUtils.failed(response.getErrMsg());
                      }
                  }
                  return RestResultUtils.success();
              } catch (Throwable ex) {
                  LOGGER.error("data import has error :", ex);
                  return RestResultUtils.failed(ex.getMessage());
              }
          });
      }
      
      @Override
      public Boolean update(List<ModifyRequest> sqlContext, BiConsumer<Boolean, Throwable> consumer) {
          try {
              
              // Since the SQL parameter is Object[], in order to ensure that the types of
              // array elements are not lost, the serialization here is done using the java-specific
              // serialization framework, rather than continuing with the protobuff
              
              LoggerUtils.printIfDebugEnabled(LOGGER, "modifyRequests info : {}", sqlContext);
              
              // {timestamp}-{group}-{ip:port}-{signature}
              
              final String key =
                      System.currentTimeMillis() + "-" + group() + "-" + memberManager.getSelf().getAddress() + "-"
                              + MD5Utils.md5Hex(sqlContext.toString(), PersistenceConstant.DEFAULT_ENCODE);
              WriteRequest request = WriteRequest.newBuilder().setGroup(group()).setKey(key)
                      .setData(ByteString.copyFrom(serializer.serialize(sqlContext)))
                      .putAllExtendInfo(EmbeddedStorageContextHolder.getCurrentExtendInfo())
                      .setType(sqlContext.getClass().getCanonicalName()).build();
              if (Objects.isNull(consumer)) {
                  Response response = this.protocol.write(request);
                  if (response.getSuccess()) {
                      return true;
                  }
                  LOGGER.error("execute sql modify operation failed : {}", response.getErrMsg());
                  return false;
              } else {
                  this.protocol.writeAsync(request).whenComplete((BiConsumer<Response, Throwable>) (response, ex) -> {
                      String errMsg = Objects.isNull(ex) ? response.getErrMsg() : ExceptionUtil.getCause(ex).getMessage();
                      consumer.accept(response.getSuccess(),
                              StringUtils.isBlank(errMsg) ? null : new NJdbcException(errMsg));
                  });
              }
              return true;
          } catch (TimeoutException e) {
              LOGGER.error("An timeout exception occurred during the update operation");
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          } catch (Throwable e) {
              LOGGER.error("An exception occurred during the update operation : {}", e);
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
          }
      }
      
      @Override
      public List<SnapshotOperation> loadSnapshotOperate() {
          return Collections.singletonList(new DerbySnapshotOperation(writeLock));
      }
      
      @SuppressWarnings("all")
      @Override
      public Response onRequest(final ReadRequest request) {
          SelectRequest selectRequest = null;
          readLock.lock();
          Object data;
          try {
              selectRequest = serializer.deserialize(request.getData().toByteArray(), SelectRequest.class);
              LoggerUtils.printIfDebugEnabled(LOGGER, "getData info : selectRequest : {}", selectRequest);
              sqlLimiter.doLimitForSelectRequest(selectRequest);
              final RowMapper<Object> mapper = RowMapperManager.getRowMapper(selectRequest.getClassName());
              final byte type = selectRequest.getQueryType();
              switch (type) {
                  case QueryType.QUERY_ONE_WITH_MAPPER_WITH_ARGS:
                      data = queryOne(jdbcTemplate, selectRequest.getSql(), selectRequest.getArgs(), mapper);
                      break;
                  case QueryType.QUERY_ONE_NO_MAPPER_NO_ARGS:
                      data = queryOne(jdbcTemplate, selectRequest.getSql(),
                              ClassUtils.findClassByName(selectRequest.getClassName()));
                      break;
                  case QueryType.QUERY_ONE_NO_MAPPER_WITH_ARGS:
                      data = queryOne(jdbcTemplate, selectRequest.getSql(), selectRequest.getArgs(),
                              ClassUtils.findClassByName(selectRequest.getClassName()));
                      break;
                  case QueryType.QUERY_MANY_WITH_MAPPER_WITH_ARGS:
                      data = queryMany(jdbcTemplate, selectRequest.getSql(), selectRequest.getArgs(), mapper);
                      break;
                  case QueryType.QUERY_MANY_WITH_LIST_WITH_ARGS:
                      data = queryMany(jdbcTemplate, selectRequest.getSql(), selectRequest.getArgs());
                      break;
                  case QueryType.QUERY_MANY_NO_MAPPER_WITH_ARGS:
                      data = queryMany(jdbcTemplate, selectRequest.getSql(), selectRequest.getArgs(),
                              ClassUtils.findClassByName(selectRequest.getClassName()));
                      break;
                  default:
                      throw new IllegalArgumentException("Unsupported data query categories");
              }
              ByteString bytes = data == null ? ByteString.EMPTY : ByteString.copyFrom(serializer.serialize(data));
              return Response.newBuilder().setSuccess(true).setData(bytes).build();
          } catch (Exception e) {
              LOGGER.error("There was an error querying the data, request : {}, error : {}", selectRequest, e.toString());
              return Response.newBuilder().setSuccess(false)
                      .setErrMsg(ClassUtils.getSimplaName(e) + ":" + ExceptionUtil.getCause(e).getMessage()).build();
          } finally {
              readLock.unlock();
          }
      }
      
      @Override
      public Response onApply(WriteRequest log) {
          LoggerUtils.printIfDebugEnabled(LOGGER, "onApply info : log : {}", log);
          final ByteString byteString = log.getData();
          Preconditions.checkArgument(byteString != null, "Log.getData() must not null");
          final Lock lock = readLock;
          lock.lock();
          try {
              List<ModifyRequest> sqlContext = serializer.deserialize(byteString.toByteArray(), List.class);
              sqlLimiter.doLimitForModifyRequest(sqlContext);
              boolean isOk = false;
              if (log.containsExtendInfo(DATA_IMPORT_KEY)) {
                  isOk = doDataImport(jdbcTemplate, sqlContext);
              } else {
                  sqlContext.sort(Comparator.comparingInt(ModifyRequest::getExecuteNo));
                  isOk = update(transactionTemplate, jdbcTemplate, sqlContext);
                  // If there is additional information, post processing
                  // Put into the asynchronous thread pool for processing to avoid blocking the
                  // normal execution of the state machine
                  PersistenceExecutor.executeEmbeddedDump(() -> {
                      for (EmbeddedApplyHook each : EmbeddedApplyHookHolder.getInstance().getAllHooks()) {
                          each.afterApply(log);
                      }
                  });
              }
              
              return Response.newBuilder().setSuccess(isOk).build();
              
              // We do not believe that an error caused by a problem with an SQL error
              // should trigger the stop operation of the raft state machine
          } catch (BadSqlGrammarException | DataIntegrityViolationException e) {
              return Response.newBuilder().setSuccess(false).setErrMsg(e.toString()).build();
          } catch (DataAccessException e) {
              throw new ConsistencyException(e.toString());
          } catch (Exception e) {
              LoggerUtils.printIfWarnEnabled(LOGGER, "onApply warn : log : {}", log, e);
              return Response.newBuilder().setSuccess(false).setErrMsg(e.toString()).build();
          } finally {
              lock.unlock();
          }
      }
      
      @Override
      public void onError(Throwable throwable) {
          // Trigger reversion strategy
          NotifyCenter.publishEvent(new RaftDbErrorEvent(throwable));
      }
      
      @Override
      public String group() {
          return PersistenceConstant.CONFIG_MODEL_RAFT_GROUP;
      }
  }
  