/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.accumulo.tserver;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Durability;
import org.apache.accumulo.core.client.SampleNotPresentException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.CompactionConfig;
import org.apache.accumulo.core.clientImpl.CompressedIterators;
import org.apache.accumulo.core.clientImpl.DurabilityImpl;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.clientImpl.TabletType;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Column;
import org.apache.accumulo.core.data.ConstraintViolationSummary;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.InitialMultiScan;
import org.apache.accumulo.core.dataImpl.thrift.InitialScan;
import org.apache.accumulo.core.dataImpl.thrift.IterInfo;
import org.apache.accumulo.core.dataImpl.thrift.MapFileInfo;
import org.apache.accumulo.core.dataImpl.thrift.MultiScanResult;
import org.apache.accumulo.core.dataImpl.thrift.ScanResult;
import org.apache.accumulo.core.dataImpl.thrift.TCMResult;
import org.apache.accumulo.core.dataImpl.thrift.TCMStatus;
import org.apache.accumulo.core.dataImpl.thrift.TColumn;
import org.apache.accumulo.core.dataImpl.thrift.TConditionalMutation;
import org.apache.accumulo.core.dataImpl.thrift.TConditionalSession;
import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.TKeyValue;
import org.apache.accumulo.core.dataImpl.thrift.TMutation;
import org.apache.accumulo.core.dataImpl.thrift.TRange;
import org.apache.accumulo.core.dataImpl.thrift.TRowRange;
import org.apache.accumulo.core.dataImpl.thrift.TSummaries;
import org.apache.accumulo.core.dataImpl.thrift.TSummaryRequest;
import org.apache.accumulo.core.dataImpl.thrift.UpdateErrors;
import org.apache.accumulo.core.iterators.IterationInterruptedException;
import org.apache.accumulo.core.logging.TabletLogger;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.TabletFile;
import org.apache.accumulo.core.sample.impl.SamplerConfigurationImpl;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
import org.apache.accumulo.core.spi.cache.BlockCache;
import org.apache.accumulo.core.spi.scan.ScanDispatcher;
import org.apache.accumulo.core.summary.Gatherer;
import org.apache.accumulo.core.summary.Gatherer.FileSystemResolver;
import org.apache.accumulo.core.summary.SummaryCollection;
import org.apache.accumulo.core.tabletserver.thrift.ActiveCompaction;
import org.apache.accumulo.core.tabletserver.thrift.ActiveScan;
import org.apache.accumulo.core.tabletserver.thrift.ConstraintViolationException;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
import org.apache.accumulo.core.tabletserver.thrift.TDurability;
import org.apache.accumulo.core.tabletserver.thrift.TSampleNotPresentException;
import org.apache.accumulo.core.tabletserver.thrift.TSamplerConfiguration;
import org.apache.accumulo.core.tabletserver.thrift.TUnloadTabletGoal;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.Halt;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.fate.zookeeper.ZooLock;
import org.apache.accumulo.fate.zookeeper.ZooUtil;
import org.apache.accumulo.server.client.ClientServiceHandler;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.data.ServerMutation;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.zookeeper.TransactionWatcher;
import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
import org.apache.accumulo.tserver.RowLocks.RowLock;
import org.apache.accumulo.tserver.data.ServerConditionalMutation;
import org.apache.accumulo.tserver.scan.LookupTask;
import org.apache.accumulo.tserver.scan.NextBatchTask;
import org.apache.accumulo.tserver.scan.ScanParameters;
import org.apache.accumulo.tserver.session.ConditionalSession;
import org.apache.accumulo.tserver.session.MultiScanSession;
import org.apache.accumulo.tserver.session.SingleScanSession;
import org.apache.accumulo.tserver.session.SummarySession;
import org.apache.accumulo.tserver.session.UpdateSession;
import org.apache.accumulo.tserver.tablet.CommitSession;
import org.apache.accumulo.tserver.tablet.CompactionInfo;
import org.apache.accumulo.tserver.tablet.Compactor;
import org.apache.accumulo.tserver.tablet.KVEntry;
import org.apache.accumulo.tserver.tablet.PreparedMutations;
import org.apache.accumulo.tserver.tablet.ScanBatch;
import org.apache.accumulo.tserver.tablet.Tablet;
import org.apache.accumulo.tserver.tablet.TabletClosedException;
import org.apache.hadoop.fs.FSError;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.apache.thrift.TException;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.google.common.collect.Collections2;

class ThriftClientHandler extends ClientServiceHandler implements TabletClientService.Iface {

  private static final Logger log = LoggerFactory.getLogger(ThriftClientHandler.class);
  private static final long MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS = 1000;
  private static final long RECENTLY_SPLIT_MILLIES = 60 * 1000;
  private final TabletServer server;
  private final WriteTracker writeTracker = new WriteTracker();
  private final RowLocks rowLocks = new RowLocks();

  ThriftClientHandler(TabletServer server) {
    super(server.getContext(), new TransactionWatcher(server.getContext()));
    this.server = server;
    log.debug("{} created", ThriftClientHandler.class.getName());
  }

  @Override
  public List<TKeyExtent> bulkImport(TInfo tinfo, TCredentials credentials, final long tid,
      final Map<TKeyExtent,Map<String,MapFileInfo>> files, final boolean setTime)
      throws ThriftSecurityException {

    if (!security.canPerformSystemActions(credentials)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }

    try {
      return transactionWatcher.run(Constants.BULK_ARBITRATOR_TYPE, tid, () -> {
        List<TKeyExtent> failures = new ArrayList<>();

        for (Entry<TKeyExtent,Map<String,MapFileInfo>> entry : files.entrySet()) {
          TKeyExtent tke = entry.getKey();
          Map<String,MapFileInfo> fileMap = entry.getValue();
          Map<TabletFile,MapFileInfo> fileRefMap = new HashMap<>();
          for (Entry<String,MapFileInfo> mapping : fileMap.entrySet()) {
            Path path = new Path(mapping.getKey());
            FileSystem ns = context.getVolumeManager().getFileSystemByPath(path);
            path = ns.makeQualified(path);
            fileRefMap.put(new TabletFile(path), mapping.getValue());
          }

          Tablet importTablet = server.getOnlineTablet(KeyExtent.fromThrift(tke));

          if (importTablet == null) {
            failures.add(tke);
          } else {
            try {
              importTablet.importMapFiles(tid, fileRefMap, setTime);
            } catch (IOException ioe) {
              log.info("files {} not imported to {}: {}", fileMap.keySet(),
                  KeyExtent.fromThrift(tke), ioe.getMessage());
              failures.add(tke);
            }
          }
        }
        return failures;
      });
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public void loadFiles(TInfo tinfo, TCredentials credentials, long tid, String dir,
      Map<TKeyExtent,Map<String,MapFileInfo>> tabletImports, boolean setTime)
      throws ThriftSecurityException {
    if (!security.canPerformSystemActions(credentials)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }

    transactionWatcher.runQuietly(Constants.BULK_ARBITRATOR_TYPE, tid, () -> {
      tabletImports.forEach((tke, fileMap) -> {
        Map<TabletFile,MapFileInfo> newFileMap = new HashMap<>();
        for (Entry<String,MapFileInfo> mapping : fileMap.entrySet()) {
          Path path = new Path(dir, mapping.getKey());
          FileSystem ns = context.getVolumeManager().getFileSystemByPath(path);
          path = ns.makeQualified(path);
          newFileMap.put(new TabletFile(path), mapping.getValue());
        }

        Tablet importTablet = server.getOnlineTablet(KeyExtent.fromThrift(tke));

        if (importTablet != null) {
          try {
            importTablet.importMapFiles(tid, newFileMap, setTime);
          } catch (IOException ioe) {
            log.debug("files {} not imported to {}: {}", fileMap.keySet(),
                KeyExtent.fromThrift(tke), ioe.getMessage());
          }
        }
      });
    });

  }

  private ScanDispatcher getScanDispatcher(KeyExtent extent) {
    if (extent.isRootTablet() || extent.isMeta()) {
      // dispatcher is only for user tables
      return null;
    }

    return context.getTableConfiguration(extent.tableId()).getScanDispatcher();
  }

  @Override
  public InitialScan startScan(TInfo tinfo, TCredentials credentials, TKeyExtent textent,
      TRange range, List<TColumn> columns, int batchSize, List<IterInfo> ssiList,
      Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites,
      boolean isolated, long readaheadThreshold, TSamplerConfiguration tSamplerConfig,
      long batchTimeOut, String contextArg, Map<String,String> executionHints)
      throws NotServingTabletException, ThriftSecurityException,
      org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException,
      TSampleNotPresentException {

    TableId tableId = TableId.of(new String(textent.getTable(), UTF_8));
    NamespaceId namespaceId;
    try {
      namespaceId = Tables.getNamespaceId(server.getContext(), tableId);
    } catch (TableNotFoundException e1) {
      throw new NotServingTabletException(textent);
    }
    if (!security.canScan(credentials, tableId, namespaceId, range, columns, ssiList, ssio,
        authorizations)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }

    if (!security.authenticatedUserHasAuthorizations(credentials, authorizations)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.BAD_AUTHORIZATIONS);
    }

    final KeyExtent extent = KeyExtent.fromThrift(textent);

    // wait for any writes that are in flight.. this done to ensure
    // consistency across client restarts... assume a client writes
    // to accumulo and dies while waiting for a confirmation from
    // accumulo... the client process restarts and tries to read
    // data from accumulo making the assumption that it will get
    // any writes previously made... however if the server side thread
    // processing the write from the dead client is still in progress,
    // the restarted client may not see the write unless we wait here.
    // this behavior is very important when the client is reading the
    // metadata
    if (waitForWrites) {
      writeTracker.waitForWrites(TabletType.type(extent));
    }

    Tablet tablet = server.getOnlineTablet(extent);
    if (tablet == null) {
      throw new NotServingTabletException(textent);
    }

    HashSet<Column> columnSet = new HashSet<>();
    for (TColumn tcolumn : columns) {
      columnSet.add(new Column(tcolumn));
    }

    ScanParameters scanParams = new ScanParameters(batchSize, new Authorizations(authorizations),
        columnSet, ssiList, ssio, isolated, SamplerConfigurationImpl.fromThrift(tSamplerConfig),
        batchTimeOut, contextArg);

    final SingleScanSession scanSession =
        new SingleScanSession(credentials, extent, scanParams, readaheadThreshold, executionHints);
    scanSession.scanner =
        tablet.createScanner(new Range(range), scanParams, scanSession.interruptFlag);

    long sid = server.sessionManager.createSession(scanSession, true);

    ScanResult scanResult;
    try {
      scanResult = continueScan(tinfo, sid, scanSession);
    } catch (NoSuchScanIDException e) {
      log.error("The impossible happened", e);
      throw new RuntimeException();
    } finally {
      server.sessionManager.unreserveSession(sid);
    }

    return new InitialScan(sid, scanResult);
  }

  @Override
  public ScanResult continueScan(TInfo tinfo, long scanID) throws NoSuchScanIDException,
      NotServingTabletException, org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException,
      TSampleNotPresentException {
    SingleScanSession scanSession =
        (SingleScanSession) server.sessionManager.reserveSession(scanID);
    if (scanSession == null) {
      throw new NoSuchScanIDException();
    }

    try {
      return continueScan(tinfo, scanID, scanSession);
    } finally {
      server.sessionManager.unreserveSession(scanSession);
    }
  }

  private ScanResult continueScan(TInfo tinfo, long scanID, SingleScanSession scanSession)
      throws NoSuchScanIDException, NotServingTabletException,
      org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException,
      TSampleNotPresentException {

    if (scanSession.nextBatchTask == null) {
      scanSession.nextBatchTask = new NextBatchTask(server, scanID, scanSession.interruptFlag);
      server.resourceManager.executeReadAhead(scanSession.extent,
          getScanDispatcher(scanSession.extent), scanSession, scanSession.nextBatchTask);
    }

    ScanBatch bresult;
    try {
      bresult = scanSession.nextBatchTask.get(MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS,
          TimeUnit.MILLISECONDS);
      scanSession.nextBatchTask = null;
    } catch (ExecutionException e) {
      server.sessionManager.removeSession(scanID);
      if (e.getCause() instanceof NotServingTabletException) {
        throw (NotServingTabletException) e.getCause();
      } else if (e.getCause() instanceof TooManyFilesException) {
        throw new org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException(
            scanSession.extent.toThrift());
      } else if (e.getCause() instanceof SampleNotPresentException) {
        throw new TSampleNotPresentException(scanSession.extent.toThrift());
      } else if (e.getCause() instanceof IOException) {
        sleepUninterruptibly(MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS, TimeUnit.MILLISECONDS);
        List<KVEntry> empty = Collections.emptyList();
        bresult = new ScanBatch(empty, true);
        scanSession.nextBatchTask = null;
      } else {
        throw new RuntimeException(e);
      }
    } catch (CancellationException ce) {
      server.sessionManager.removeSession(scanID);
      Tablet tablet = server.getOnlineTablet(scanSession.extent);
      if (tablet == null || tablet.isClosed()) {
        throw new NotServingTabletException(scanSession.extent.toThrift());
      } else {
        throw new NoSuchScanIDException();
      }
    } catch (TimeoutException e) {
      List<TKeyValue> param = Collections.emptyList();
      long timeout = server.getConfiguration().getTimeInMillis(Property.TSERV_CLIENT_TIMEOUT);
      server.sessionManager.removeIfNotAccessed(scanID, timeout);
      return new ScanResult(param, true);
    } catch (Exception t) {
      server.sessionManager.removeSession(scanID);
      log.warn("Failed to get next batch", t);
      throw new RuntimeException(t);
    }

    ScanResult scanResult = new ScanResult(Key.compress(bresult.getResults()), bresult.isMore());

    scanSession.entriesReturned += scanResult.results.size();

    scanSession.batchCount++;

    if (scanResult.more && scanSession.batchCount > scanSession.readaheadThreshold) {
      // start reading next batch while current batch is transmitted
      // to client
      scanSession.nextBatchTask = new NextBatchTask(server, scanID, scanSession.interruptFlag);
      server.resourceManager.executeReadAhead(scanSession.extent,
          getScanDispatcher(scanSession.extent), scanSession, scanSession.nextBatchTask);
    }

    if (!scanResult.more) {
      closeScan(tinfo, scanID);
    }

    return scanResult;
  }

  @Override
  public void closeScan(TInfo tinfo, long scanID) {
    final SingleScanSession ss = (SingleScanSession) server.sessionManager.removeSession(scanID);
    if (ss != null) {
      long t2 = System.currentTimeMillis();

      if (log.isTraceEnabled()) {
        log.trace(String.format("ScanSess tid %s %s %,d entries in %.2f secs, nbTimes = [%s] ",
            TServerUtils.clientAddress.get(), ss.extent.tableId(), ss.entriesReturned,
            (t2 - ss.startTime) / 1000.0, ss.runStats.toString()));
      }

      server.scanMetrics.addScan(t2 - ss.startTime);
      server.scanMetrics.addResult(ss.entriesReturned);
    }
  }

  @Override
  public InitialMultiScan startMultiScan(TInfo tinfo, TCredentials credentials,
      Map<TKeyExtent,List<TRange>> tbatch, List<TColumn> tcolumns, List<IterInfo> ssiList,
      Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites,
      TSamplerConfiguration tSamplerConfig, long batchTimeOut, String contextArg,
      Map<String,String> executionHints)
      throws ThriftSecurityException, TSampleNotPresentException {
    // find all of the tables that need to be scanned
    final HashSet<TableId> tables = new HashSet<>();
    for (TKeyExtent keyExtent : tbatch.keySet()) {
      tables.add(TableId.of(new String(keyExtent.getTable(), UTF_8)));
    }

    if (tables.size() != 1) {
      throw new IllegalArgumentException("Cannot batch scan over multiple tables");
    }

    // check if user has permission to the tables
    for (TableId tableId : tables) {
      NamespaceId namespaceId = getNamespaceId(credentials, tableId);
      if (!security.canScan(credentials, tableId, namespaceId, tbatch, tcolumns, ssiList, ssio,
          authorizations)) {
        throw new ThriftSecurityException(credentials.getPrincipal(),
            SecurityErrorCode.PERMISSION_DENIED);
      }
    }

    try {
      if (!security.authenticatedUserHasAuthorizations(credentials, authorizations)) {
        throw new ThriftSecurityException(credentials.getPrincipal(),
            SecurityErrorCode.BAD_AUTHORIZATIONS);
      }
    } catch (ThriftSecurityException tse) {
      log.error("{} is not authorized", credentials.getPrincipal(), tse);
      throw tse;
    }

    // @formatter:off
    Map<KeyExtent, List<Range>> batch = tbatch.entrySet().stream().collect(Collectors.toMap(
                    entry -> KeyExtent.fromThrift(entry.getKey()),
                    entry -> entry.getValue().stream().map(Range::new).collect(Collectors.toList())
    ));
    // @formatter:on

    // This is used to determine which thread pool to use
    KeyExtent threadPoolExtent = batch.keySet().iterator().next();

    if (waitForWrites) {
      writeTracker.waitForWrites(TabletType.type(batch.keySet()));
    }

    Set<Column> columnSet = tcolumns.isEmpty() ? Collections.emptySet()
        : new HashSet<>(Collections2.transform(tcolumns, Column::new));

    ScanParameters scanParams =
        new ScanParameters(-1, new Authorizations(authorizations), columnSet, ssiList, ssio, false,
            SamplerConfigurationImpl.fromThrift(tSamplerConfig), batchTimeOut, contextArg);

    final MultiScanSession mss =
        new MultiScanSession(credentials, threadPoolExtent, batch, scanParams, executionHints);

    mss.numTablets = batch.size();
    for (List<Range> ranges : batch.values()) {
      mss.numRanges += ranges.size();
    }

    long sid = server.sessionManager.createSession(mss, true);

    MultiScanResult result;
    try {
      result = continueMultiScan(sid, mss);
    } finally {
      server.sessionManager.unreserveSession(sid);
    }

    return new InitialMultiScan(sid, result);
  }

  @Override
  public MultiScanResult continueMultiScan(TInfo tinfo, long scanID)
      throws NoSuchScanIDException, TSampleNotPresentException {

    MultiScanSession session = (MultiScanSession) server.sessionManager.reserveSession(scanID);

    if (session == null) {
      throw new NoSuchScanIDException();
    }

    try {
      return continueMultiScan(scanID, session);
    } finally {
      server.sessionManager.unreserveSession(session);
    }
  }

  private MultiScanResult continueMultiScan(long scanID, MultiScanSession session)
      throws TSampleNotPresentException {

    if (session.lookupTask == null) {
      session.lookupTask = new LookupTask(server, scanID);
      server.resourceManager.executeReadAhead(session.threadPoolExtent,
          getScanDispatcher(session.threadPoolExtent), session, session.lookupTask);
    }

    try {
      MultiScanResult scanResult =
          session.lookupTask.get(MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS, TimeUnit.MILLISECONDS);
      session.lookupTask = null;
      return scanResult;
    } catch (ExecutionException e) {
      server.sessionManager.removeSession(scanID);
      if (e.getCause() instanceof SampleNotPresentException) {
        throw new TSampleNotPresentException();
      } else {
        log.warn("Failed to get multiscan result", e);
        throw new RuntimeException(e);
      }
    } catch (TimeoutException e1) {
      long timeout = server.getConfiguration().getTimeInMillis(Property.TSERV_CLIENT_TIMEOUT);
      server.sessionManager.removeIfNotAccessed(scanID, timeout);
      List<TKeyValue> results = Collections.emptyList();
      Map<TKeyExtent,List<TRange>> failures = Collections.emptyMap();
      List<TKeyExtent> fullScans = Collections.emptyList();
      return new MultiScanResult(results, failures, fullScans, null, null, false, true);
    } catch (Exception t) {
      server.sessionManager.removeSession(scanID);
      log.warn("Failed to get multiscan result", t);
      throw new RuntimeException(t);
    }
  }

  @Override
  public void closeMultiScan(TInfo tinfo, long scanID) throws NoSuchScanIDException {
    MultiScanSession session = (MultiScanSession) server.sessionManager.removeSession(scanID);
    if (session == null) {
      throw new NoSuchScanIDException();
    }

    long t2 = System.currentTimeMillis();

    if (log.isTraceEnabled()) {
      log.trace(String.format(
          "MultiScanSess %s %,d entries in %.2f secs"
              + " (lookup_time:%.2f secs tablets:%,d ranges:%,d) ",
          TServerUtils.clientAddress.get(), session.numEntries, (t2 - session.startTime) / 1000.0,
          session.totalLookupTime / 1000.0, session.numTablets, session.numRanges));
    }
  }

  @Override
  public long startUpdate(TInfo tinfo, TCredentials credentials, TDurability tdurabilty)
      throws ThriftSecurityException {
    // Make sure user is real
    Durability durability = DurabilityImpl.fromThrift(tdurabilty);
    security.authenticateUser(credentials, credentials);
    server.updateMetrics.addPermissionErrors(0);

    UpdateSession us =
        new UpdateSession(new TservConstraintEnv(server.getContext(), security, credentials),
            credentials, durability);
    return server.sessionManager.createSession(us, false);
  }

  private void setUpdateTablet(UpdateSession us, KeyExtent keyExtent) {
    long t1 = System.currentTimeMillis();
    if (us.currentTablet != null && us.currentTablet.getExtent().equals(keyExtent)) {
      return;
    }
    if (us.currentTablet == null
        && (us.failures.containsKey(keyExtent) || us.authFailures.containsKey(keyExtent))) {
      // if there were previous failures, then do not accept additional writes
      return;
    }

    TableId tableId = null;
    try {
      // if user has no permission to write to this table, add it to
      // the failures list
      boolean sameTable = us.currentTablet != null
          && (us.currentTablet.getExtent().tableId().equals(keyExtent.tableId()));
      tableId = keyExtent.tableId();
      if (sameTable || security.canWrite(us.getCredentials(), tableId,
          Tables.getNamespaceId(server.getContext(), tableId))) {
        long t2 = System.currentTimeMillis();
        us.authTimes.addStat(t2 - t1);
        us.currentTablet = server.getOnlineTablet(keyExtent);
        if (us.currentTablet != null) {
          us.queuedMutations.put(us.currentTablet, new ArrayList<>());
        } else {
          // not serving tablet, so report all mutations as
          // failures
          us.failures.put(keyExtent, 0L);
          server.updateMetrics.addUnknownTabletErrors(0);
        }
      } else {
        log.warn("Denying access to table {} for user {}", keyExtent.tableId(), us.getUser());
        long t2 = System.currentTimeMillis();
        us.authTimes.addStat(t2 - t1);
        us.currentTablet = null;
        us.authFailures.put(keyExtent, SecurityErrorCode.PERMISSION_DENIED);
        server.updateMetrics.addPermissionErrors(0);
        return;
      }
    } catch (TableNotFoundException tnfe) {
      log.error("Table " + tableId + " not found ", tnfe);
      long t2 = System.currentTimeMillis();
      us.authTimes.addStat(t2 - t1);
      us.currentTablet = null;
      us.authFailures.put(keyExtent, SecurityErrorCode.TABLE_DOESNT_EXIST);
      server.updateMetrics.addUnknownTabletErrors(0);
      return;
    } catch (ThriftSecurityException e) {
      log.error("Denying permission to check user " + us.getUser() + " with user " + e.getUser(),
          e);
      long t2 = System.currentTimeMillis();
      us.authTimes.addStat(t2 - t1);
      us.currentTablet = null;
      us.authFailures.put(keyExtent, e.getCode());
      server.updateMetrics.addPermissionErrors(0);
      return;
    }
  }

  @Override
  public void applyUpdates(TInfo tinfo, long updateID, TKeyExtent tkeyExtent,
      List<TMutation> tmutations) {
    UpdateSession us = (UpdateSession) server.sessionManager.reserveSession(updateID);
    if (us == null) {
      return;
    }

    boolean reserved = true;
    try {
      KeyExtent keyExtent = KeyExtent.fromThrift(tkeyExtent);
      setUpdateTablet(us, keyExtent);

      if (us.currentTablet != null) {
        long additionalMutationSize = 0;
        List<Mutation> mutations = us.queuedMutations.get(us.currentTablet);
        for (TMutation tmutation : tmutations) {
          Mutation mutation = new ServerMutation(tmutation);
          mutations.add(mutation);
          additionalMutationSize += mutation.numBytes();
        }
        us.queuedMutationSize += additionalMutationSize;
        long totalQueued = server.updateTotalQueuedMutationSize(additionalMutationSize);
        long total = server.getConfiguration().getAsBytes(Property.TSERV_TOTAL_MUTATION_QUEUE_MAX);
        if (totalQueued > total) {
          try {
            flush(us);
          } catch (HoldTimeoutException hte) {
            // Assumption is that the client has timed out and is gone. If that's not the case,
            // then removing the session should cause the client to fail
            // in such a way that it retries.
            log.debug("HoldTimeoutException during applyUpdates, removing session");
            server.sessionManager.removeSession(updateID, true);
            reserved = false;
          }
        }
      }
    } finally {
      if (reserved) {
        server.sessionManager.unreserveSession(us);
      }
    }
  }

  private void flush(UpdateSession us) {

    int mutationCount = 0;
    Map<CommitSession,List<Mutation>> sendables = new HashMap<>();
    Map<CommitSession,TabletMutations> loggables = new HashMap<>();
    Throwable error = null;

    long pt1 = System.currentTimeMillis();

    boolean containsMetadataTablet = false;
    for (Tablet tablet : us.queuedMutations.keySet()) {
      if (tablet.getExtent().isMeta()) {
        containsMetadataTablet = true;
      }
    }

    if (!containsMetadataTablet && !us.queuedMutations.isEmpty()) {
      server.resourceManager.waitUntilCommitsAreEnabled();
    }

    try (TraceScope prep = Trace.startSpan("prep")) {
      for (Entry<Tablet,? extends List<Mutation>> entry : us.queuedMutations.entrySet()) {

        Tablet tablet = entry.getKey();
        Durability durability =
            DurabilityImpl.resolveDurabilty(us.durability, tablet.getDurability());
        List<Mutation> mutations = entry.getValue();
        if (!mutations.isEmpty()) {
          try {
            server.updateMetrics.addMutationArraySize(mutations.size());

            PreparedMutations prepared = tablet.prepareMutationsForCommit(us.cenv, mutations);

            if (prepared.tabletClosed()) {
              if (us.currentTablet == tablet) {
                us.currentTablet = null;
              }
              us.failures.put(tablet.getExtent(), us.successfulCommits.get(tablet));
            } else {
              if (!prepared.getNonViolators().isEmpty()) {
                List<Mutation> validMutations = prepared.getNonViolators();
                CommitSession session = prepared.getCommitSession();
                if (durability != Durability.NONE) {
                  loggables.put(session, new TabletMutations(session, validMutations, durability));
                }
                sendables.put(session, validMutations);
              }

              if (!prepared.getViolations().isEmpty()) {
                us.violations.add(prepared.getViolations());
                server.updateMetrics.addConstraintViolations(0);
              }
              // Use the size of the original mutation list, regardless of how many mutations
              // did not violate constraints.
              mutationCount += mutations.size();

            }
          } catch (Exception t) {
            error = t;
            log.error("Unexpected error preparing for commit", error);
            break;
          }
        }
      }
    }

    long pt2 = System.currentTimeMillis();
    us.prepareTimes.addStat(pt2 - pt1);
    updateAvgPrepTime(pt2 - pt1, us.queuedMutations.size());

    if (error != null) {
      sendables.forEach((commitSession, value) -> commitSession.abortCommit());
      throw new RuntimeException(error);
    }
    try {
      try (TraceScope wal = Trace.startSpan("wal")) {
        while (true) {
          try {
            long t1 = System.currentTimeMillis();

            server.logger.logManyTablets(loggables);

            long t2 = System.currentTimeMillis();
            us.walogTimes.addStat(t2 - t1);
            updateWalogWriteTime((t2 - t1));
            break;
          } catch (IOException | FSError ex) {
            log.warn("logging mutations failed, retrying");
          } catch (Exception t) {
            log.error("Unknown exception logging mutations, counts"
                + " for mutations in flight not decremented!", t);
            throw new RuntimeException(t);
          }
        }
      }

      try (TraceScope commit = Trace.startSpan("commit")) {
        long t1 = System.currentTimeMillis();
        sendables.forEach((commitSession, mutations) -> {
          commitSession.commit(mutations);
          KeyExtent extent = commitSession.getExtent();

          if (us.currentTablet != null && extent == us.currentTablet.getExtent()) {
            // because constraint violations may filter out some
            // mutations, for proper accounting with the client code,
            // need to increment the count based on the original
            // number of mutations from the client NOT the filtered number
            us.successfulCommits.increment(us.currentTablet,
                us.queuedMutations.get(us.currentTablet).size());
          }
        });
        long t2 = System.currentTimeMillis();

        us.flushTime += (t2 - pt1);
        us.commitTimes.addStat(t2 - t1);

        updateAvgCommitTime(t2 - t1, sendables.size());
      }
    } finally {
      us.queuedMutations.clear();
      if (us.currentTablet != null) {
        us.queuedMutations.put(us.currentTablet, new ArrayList<>());
      }
      server.updateTotalQueuedMutationSize(-us.queuedMutationSize);
      us.queuedMutationSize = 0;
    }
    us.totalUpdates += mutationCount;
  }

  private void updateWalogWriteTime(long time) {
    server.updateMetrics.addWalogWriteTime(time);
  }

  private void updateAvgCommitTime(long time, int size) {
    if (size > 0)
      server.updateMetrics.addCommitTime((long) (time / (double) size));
  }

  private void updateAvgPrepTime(long time, int size) {
    if (size > 0)
      server.updateMetrics.addCommitPrep((long) (time / (double) size));
  }

  @Override
  public UpdateErrors closeUpdate(TInfo tinfo, long updateID) throws NoSuchScanIDException {
    final UpdateSession us = (UpdateSession) server.sessionManager.removeSession(updateID);
    if (us == null) {
      throw new NoSuchScanIDException();
    }

    // clients may or may not see data from an update session while
    // it is in progress, however when the update session is closed
    // want to ensure that reads wait for the write to finish
    long opid = writeTracker.startWrite(us.queuedMutations.keySet());

    try {
      flush(us);
    } catch (HoldTimeoutException e) {
      // Assumption is that the client has timed out and is gone. If that's not the case throw an
      // exception that will cause it to retry.
      log.debug("HoldTimeoutException during closeUpdate, reporting no such session");
      throw new NoSuchScanIDException();
    } finally {
      writeTracker.finishWrite(opid);
    }

    if (log.isTraceEnabled()) {
      log.trace(
          String.format("UpSess %s %,d in %.3fs, at=[%s] ft=%.3fs(pt=%.3fs lt=%.3fs ct=%.3fs)",
              TServerUtils.clientAddress.get(), us.totalUpdates,
              (System.currentTimeMillis() - us.startTime) / 1000.0, us.authTimes.toString(),
              us.flushTime / 1000.0, us.prepareTimes.sum() / 1000.0, us.walogTimes.sum() / 1000.0,
              us.commitTimes.sum() / 1000.0));
    }
    if (!us.failures.isEmpty()) {
      Entry<KeyExtent,Long> first = us.failures.entrySet().iterator().next();
      log.debug(String.format("Failures: %d, first extent %s successful commits: %d",
          us.failures.size(), first.getKey().toString(), first.getValue()));
    }
    List<ConstraintViolationSummary> violations = us.violations.asList();
    if (!violations.isEmpty()) {
      ConstraintViolationSummary first = us.violations.asList().iterator().next();
      log.debug(String.format("Violations: %d, first %s occurs %d", violations.size(),
          first.violationDescription, first.numberOfViolatingMutations));
    }
    if (!us.authFailures.isEmpty()) {
      KeyExtent first = us.authFailures.keySet().iterator().next();
      log.debug(String.format("Authentication Failures: %d, first %s", us.authFailures.size(),
          first.toString()));
    }
    return new UpdateErrors(
        us.failures.entrySet().stream()
            .collect(Collectors.toMap(e -> e.getKey().toThrift(), Entry::getValue)),
        violations.stream().map(ConstraintViolationSummary::toThrift).collect(Collectors.toList()),
        us.authFailures.entrySet().stream()
            .collect(Collectors.toMap(e -> e.getKey().toThrift(), Entry::getValue)));
  }

  @Override
  public void update(TInfo tinfo, TCredentials credentials, TKeyExtent tkeyExtent,
      TMutation tmutation, TDurability tdurability)
      throws NotServingTabletException, ConstraintViolationException, ThriftSecurityException {

    final TableId tableId = TableId.of(new String(tkeyExtent.getTable(), UTF_8));
    NamespaceId namespaceId = getNamespaceId(credentials, tableId);
    if (!security.canWrite(credentials, tableId, namespaceId)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }
    final KeyExtent keyExtent = KeyExtent.fromThrift(tkeyExtent);
    final Tablet tablet = server.getOnlineTablet(KeyExtent.copyOf(keyExtent));
    if (tablet == null) {
      throw new NotServingTabletException(tkeyExtent);
    }
    Durability tabletDurability = tablet.getDurability();

    if (!keyExtent.isMeta()) {
      try {
        server.resourceManager.waitUntilCommitsAreEnabled();
      } catch (HoldTimeoutException hte) {
        // Major hack. Assumption is that the client has timed out and is gone. If that's not the
        // case, then throwing the following will let client know there
        // was a failure and it should retry.
        throw new NotServingTabletException(tkeyExtent);
      }
    }

    final long opid = writeTracker.startWrite(TabletType.type(keyExtent));

    try {
      final Mutation mutation = new ServerMutation(tmutation);
      final List<Mutation> mutations = Collections.singletonList(mutation);

      PreparedMutations prepared;
      try (TraceScope prep = Trace.startSpan("prep")) {
        prepared = tablet.prepareMutationsForCommit(
            new TservConstraintEnv(server.getContext(), security, credentials), mutations);
      }

      if (prepared.tabletClosed()) {
        throw new NotServingTabletException(tkeyExtent);
      } else if (!prepared.getViolators().isEmpty()) {
        throw new ConstraintViolationException(prepared.getViolations().asList().stream()
            .map(ConstraintViolationSummary::toThrift).collect(Collectors.toList()));
      } else {
        CommitSession session = prepared.getCommitSession();
        Durability durability = DurabilityImpl
            .resolveDurabilty(DurabilityImpl.fromThrift(tdurability), tabletDurability);

        // Instead of always looping on true, skip completely when durability is NONE.
        while (durability != Durability.NONE) {
          try {
            try (TraceScope wal = Trace.startSpan("wal")) {
              server.logger.log(session, mutation, durability);
            }
            break;
          } catch (IOException ex) {
            log.warn("Error writing mutations to log", ex);
          }
        }

        try (TraceScope commit = Trace.startSpan("commit")) {
          session.commit(mutations);
        }
      }
    } finally {
      writeTracker.finishWrite(opid);
    }
  }

  private NamespaceId getNamespaceId(TCredentials credentials, TableId tableId)
      throws ThriftSecurityException {
    try {
      return Tables.getNamespaceId(server.getContext(), tableId);
    } catch (TableNotFoundException e1) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.TABLE_DOESNT_EXIST);
    }
  }

  private void checkConditions(Map<KeyExtent,List<ServerConditionalMutation>> updates,
      ArrayList<TCMResult> results, ConditionalSession cs, List<String> symbols)
      throws IOException {
    Iterator<Entry<KeyExtent,List<ServerConditionalMutation>>> iter = updates.entrySet().iterator();

    final CompressedIterators compressedIters = new CompressedIterators(symbols);
    ConditionCheckerContext checkerContext = new ConditionCheckerContext(server.getContext(),
        compressedIters, context.getTableConfiguration(cs.tableId));

    while (iter.hasNext()) {
      final Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
      final Tablet tablet = server.getOnlineTablet(entry.getKey());

      if (tablet == null || tablet.isClosed()) {
        for (ServerConditionalMutation scm : entry.getValue()) {
          results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
        }
        iter.remove();
      } else {
        final List<ServerConditionalMutation> okMutations =
            new ArrayList<>(entry.getValue().size());
        final List<TCMResult> resultsSubList = results.subList(results.size(), results.size());

        ConditionChecker checker =
            checkerContext.newChecker(entry.getValue(), okMutations, resultsSubList);
        try {
          tablet.checkConditions(checker, cs.auths, cs.interruptFlag);

          if (okMutations.isEmpty()) {
            iter.remove();
          } else {
            entry.setValue(okMutations);
          }
        } catch (TabletClosedException | IterationInterruptedException | TooManyFilesException e) {
          // clear anything added while checking conditions.
          resultsSubList.clear();

          for (ServerConditionalMutation scm : entry.getValue()) {
            results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
          }
          iter.remove();
        }
      }
    }
  }

  private void writeConditionalMutations(Map<KeyExtent,List<ServerConditionalMutation>> updates,
      ArrayList<TCMResult> results, ConditionalSession sess) {
    Set<Entry<KeyExtent,List<ServerConditionalMutation>>> es = updates.entrySet();

    Map<CommitSession,List<Mutation>> sendables = new HashMap<>();
    Map<CommitSession,TabletMutations> loggables = new HashMap<>();

    boolean sessionCanceled = sess.interruptFlag.get();

    try (TraceScope prepSpan = Trace.startSpan("prep")) {
      long t1 = System.currentTimeMillis();
      for (Entry<KeyExtent,List<ServerConditionalMutation>> entry : es) {
        final Tablet tablet = server.getOnlineTablet(entry.getKey());
        if (tablet == null || tablet.isClosed() || sessionCanceled) {
          addMutationsAsTCMResults(results, entry.getValue(), TCMStatus.IGNORED);
        } else {
          final Durability durability =
              DurabilityImpl.resolveDurabilty(sess.durability, tablet.getDurability());

          @SuppressWarnings("unchecked")
          List<Mutation> mutations = (List<Mutation>) (List<? extends Mutation>) entry.getValue();
          if (!mutations.isEmpty()) {

            PreparedMutations prepared = tablet.prepareMutationsForCommit(
                new TservConstraintEnv(server.getContext(), security, sess.credentials), mutations);

            if (prepared.tabletClosed()) {
              addMutationsAsTCMResults(results, mutations, TCMStatus.IGNORED);
            } else {
              if (!prepared.getNonViolators().isEmpty()) {
                // Only log and commit mutations that did not violate constraints.
                List<Mutation> validMutations = prepared.getNonViolators();
                addMutationsAsTCMResults(results, validMutations, TCMStatus.ACCEPTED);
                CommitSession session = prepared.getCommitSession();
                if (durability != Durability.NONE) {
                  loggables.put(session, new TabletMutations(session, validMutations, durability));
                }
                sendables.put(session, validMutations);
              }

              if (!prepared.getViolators().isEmpty()) {
                addMutationsAsTCMResults(results, prepared.getViolators(), TCMStatus.VIOLATED);
              }
            }
          }
        }
      }

      long t2 = System.currentTimeMillis();
      updateAvgPrepTime(t2 - t1, es.size());
    }

    try (TraceScope walSpan = Trace.startSpan("wal")) {
      while (!loggables.isEmpty()) {
        try {
          long t1 = System.currentTimeMillis();
          server.logger.logManyTablets(loggables);
          long t2 = System.currentTimeMillis();
          updateWalogWriteTime(t2 - t1);
          break;
        } catch (IOException | FSError ex) {
          log.warn("logging mutations failed, retrying");
        } catch (Exception t) {
          log.error("Unknown exception logging mutations, counts for"
              + " mutations in flight not decremented!", t);
          throw new RuntimeException(t);
        }
      }
    }

    try (TraceScope commitSpan = Trace.startSpan("commit")) {
      long t1 = System.currentTimeMillis();
      sendables.forEach(CommitSession::commit);
      long t2 = System.currentTimeMillis();
      updateAvgCommitTime(t2 - t1, sendables.size());
    }
  }

  /**
   * Transform and add each mutation as a {@link TCMResult} with the mutation's ID and the specified
   * status to the {@link TCMResult} list.
   */
  private void addMutationsAsTCMResults(final List<TCMResult> list,
      final Collection<? extends Mutation> mutations, final TCMStatus status) {
    mutations.stream()
        .map(mutation -> new TCMResult(((ServerConditionalMutation) mutation).getID(), status))
        .forEach(list::add);
  }

  private Map<KeyExtent,List<ServerConditionalMutation>> conditionalUpdate(ConditionalSession cs,
      Map<KeyExtent,List<ServerConditionalMutation>> updates, ArrayList<TCMResult> results,
      List<String> symbols) throws IOException {
    // sort each list of mutations, this is done to avoid deadlock and doing seeks in order is
    // more efficient and detect duplicate rows.
    ConditionalMutationSet.sortConditionalMutations(updates);

    Map<KeyExtent,List<ServerConditionalMutation>> deferred = new HashMap<>();

    // can not process two mutations for the same row, because one will not see what the other
    // writes
    ConditionalMutationSet.deferDuplicatesRows(updates, deferred);

    // get as many locks as possible w/o blocking... defer any rows that are locked
    List<RowLock> locks = rowLocks.acquireRowlocks(updates, deferred);
    try {
      try (TraceScope checkSpan = Trace.startSpan("Check conditions")) {
        checkConditions(updates, results, cs, symbols);
      }

      try (TraceScope updateSpan = Trace.startSpan("apply conditional mutations")) {
        writeConditionalMutations(updates, results, cs);
      }
    } finally {
      rowLocks.releaseRowLocks(locks);
    }
    return deferred;
  }

  @Override
  public TConditionalSession startConditionalUpdate(TInfo tinfo, TCredentials credentials,
      List<ByteBuffer> authorizations, String tableIdStr, TDurability tdurabilty,
      String classLoaderContext) throws ThriftSecurityException, TException {

    TableId tableId = TableId.of(tableIdStr);
    Authorizations userauths = null;
    NamespaceId namespaceId = getNamespaceId(credentials, tableId);
    if (!security.canConditionallyUpdate(credentials, tableId, namespaceId)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }

    userauths = security.getUserAuthorizations(credentials);
    for (ByteBuffer auth : authorizations) {
      if (!userauths.contains(ByteBufferUtil.toBytes(auth))) {
        throw new ThriftSecurityException(credentials.getPrincipal(),
            SecurityErrorCode.BAD_AUTHORIZATIONS);
      }
    }

    ConditionalSession cs = new ConditionalSession(credentials, new Authorizations(authorizations),
        tableId, DurabilityImpl.fromThrift(tdurabilty));

    long sid = server.sessionManager.createSession(cs, false);
    return new TConditionalSession(sid, server.getLockID(), server.sessionManager.getMaxIdleTime());
  }

  @Override
  public List<TCMResult> conditionalUpdate(TInfo tinfo, long sessID,
      Map<TKeyExtent,List<TConditionalMutation>> mutations, List<String> symbols)
      throws NoSuchScanIDException, TException {

    ConditionalSession cs = (ConditionalSession) server.sessionManager.reserveSession(sessID);

    if (cs == null || cs.interruptFlag.get()) {
      throw new NoSuchScanIDException();
    }

    if (!cs.tableId.equals(MetadataTable.ID) && !cs.tableId.equals(RootTable.ID)) {
      try {
        server.resourceManager.waitUntilCommitsAreEnabled();
      } catch (HoldTimeoutException hte) {
        // Assumption is that the client has timed out and is gone. If that's not the case throw
        // an exception that will cause it to retry.
        log.debug("HoldTimeoutException during conditionalUpdate, reporting no such session");
        throw new NoSuchScanIDException();
      }
    }

    TableId tid = cs.tableId;
    long opid = writeTracker.startWrite(TabletType.type(new KeyExtent(tid, null, null)));

    try {
      // @formatter:off
      Map<KeyExtent, List<ServerConditionalMutation>> updates = mutations.entrySet().stream().collect(Collectors.toMap(
                      entry -> KeyExtent.fromThrift(entry.getKey()),
                      entry -> entry.getValue().stream().map(ServerConditionalMutation::new).collect(Collectors.toList())
      ));
      // @formatter:on
      for (KeyExtent ke : updates.keySet()) {
        if (!ke.tableId().equals(tid)) {
          throw new IllegalArgumentException("Unexpected table id " + tid + " != " + ke.tableId());
        }
      }

      ArrayList<TCMResult> results = new ArrayList<>();

      Map<KeyExtent,List<ServerConditionalMutation>> deferred =
          conditionalUpdate(cs, updates, results, symbols);

      while (!deferred.isEmpty()) {
        deferred = conditionalUpdate(cs, deferred, results, symbols);
      }

      return results;
    } catch (IOException ioe) {
      throw new TException(ioe);
    } finally {
      writeTracker.finishWrite(opid);
      server.sessionManager.unreserveSession(sessID);
    }
  }

  @Override
  public void invalidateConditionalUpdate(TInfo tinfo, long sessID) {
    // this method should wait for any running conditional update to complete
    // after this method returns a conditional update should not be able to start

    ConditionalSession cs = (ConditionalSession) server.sessionManager.getSession(sessID);
    if (cs != null) {
      cs.interruptFlag.set(true);
    }

    cs = (ConditionalSession) server.sessionManager.reserveSession(sessID, true);
    if (cs != null) {
      server.sessionManager.removeSession(sessID, true);
    }
  }

  @Override
  public void closeConditionalUpdate(TInfo tinfo, long sessID) {
    server.sessionManager.removeSession(sessID, false);
  }

  @Override
  public void splitTablet(TInfo tinfo, TCredentials credentials, TKeyExtent tkeyExtent,
      ByteBuffer splitPoint) throws NotServingTabletException, ThriftSecurityException {

    TableId tableId = TableId.of(new String(ByteBufferUtil.toBytes(tkeyExtent.table)));
    NamespaceId namespaceId = getNamespaceId(credentials, tableId);

    if (!security.canSplitTablet(credentials, tableId, namespaceId)) {
      throw new ThriftSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED);
    }

    KeyExtent keyExtent = KeyExtent.fromThrift(tkeyExtent);

    Tablet tablet = server.getOnlineTablet(keyExtent);
    if (tablet == null) {
      throw new NotServingTabletException(tkeyExtent);
    }

    if (keyExtent.endRow() == null
        || !keyExtent.endRow().equals(ByteBufferUtil.toText(splitPoint))) {
      try {
        if (server.splitTablet(tablet, ByteBufferUtil.toBytes(splitPoint)) == null) {
          throw new NotServingTabletException(tkeyExtent);
        }
      } catch (IOException e) {
        log.warn("Failed to split " + keyExtent, e);
        throw new RuntimeException(e);
      }
    }
  }

  @Override
  public TabletServerStatus getTabletServerStatus(TInfo tinfo, TCredentials credentials) {
    return server.getStats(server.sessionManager.getActiveScansPerTable());
  }

  @Override
  public List<TabletStats> getTabletStats(TInfo tinfo, TCredentials credentials, String tableId) {
    List<TabletStats> result = new ArrayList<>();
    TableId text = TableId.of(tableId);
    KeyExtent start = new KeyExtent(text, new Text(), null);
    for (Entry<KeyExtent,Tablet> entry : server.getOnlineTablets().tailMap(start).entrySet()) {
      KeyExtent ke = entry.getKey();
      if (ke.tableId().compareTo(text) == 0) {
        Tablet tablet = entry.getValue();
        TabletStats stats = tablet.getTabletStats();
        stats.extent = ke.toThrift();
        stats.ingestRate = tablet.ingestRate();
        stats.queryRate = tablet.queryRate();
        stats.splitCreationTime = tablet.getSplitCreationTime();
        stats.numEntries = tablet.getNumEntries();
        result.add(stats);
      }
    }
    return result;
  }

  private void checkPermission(TCredentials credentials, String lock, final String request)
      throws ThriftSecurityException {
    try {
      log.trace("Got {} message from user: {}", request, credentials.getPrincipal());
      if (!security.canPerformSystemActions(credentials)) {
        log.warn("Got {} message from user: {}", request, credentials.getPrincipal());
        throw new ThriftSecurityException(credentials.getPrincipal(),
            SecurityErrorCode.PERMISSION_DENIED);
      }
    } catch (ThriftSecurityException e) {
      log.warn("Got {} message from unauthenticatable user: {}", request, e.getUser());
      if (context.getCredentials().getToken().getClass().getName()
          .equals(credentials.getTokenClassName())) {
        log.error("Got message from a service with a mismatched configuration."
            + " Please ensure a compatible configuration.", e);
      }
      throw e;
    }

    if (server.getLock() == null || !server.getLock().wasLockAcquired()) {
      log.debug("Got {} message before my lock was acquired, ignoring...", request);
      throw new RuntimeException("Lock not acquired");
    }

    if (server.getLock() != null && server.getLock().wasLockAcquired()
        && !server.getLock().isLocked()) {
      Halt.halt(1, () -> {
        log.info("Tablet server no longer holds lock during checkPermission() : {}, exiting",
            request);
        server.gcLogger.logGCInfo(server.getConfiguration());
      });
    }

    if (lock != null) {
      ZooUtil.LockID lid =
          new ZooUtil.LockID(context.getZooKeeperRoot() + Constants.ZMASTER_LOCK, lock);

      try {
        if (!ZooLock.isLockHeld(server.masterLockCache, lid)) {
          // maybe the cache is out of date and a new master holds the
          // lock?
          server.masterLockCache.clear();
          if (!ZooLock.isLockHeld(server.masterLockCache, lid)) {
            log.warn("Got {} message from a master that does not hold the current lock {}", request,
                lock);
            throw new RuntimeException("bad master lock");
          }
        }
      } catch (Exception e) {
        throw new RuntimeException("bad master lock", e);
      }
    }
  }

  @Override
  public void loadTablet(TInfo tinfo, TCredentials credentials, String lock,
      final TKeyExtent textent) {

    try {
      checkPermission(credentials, lock, "loadTablet");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to load a tablet", e);
      throw new RuntimeException(e);
    }

    final KeyExtent extent = KeyExtent.fromThrift(textent);

    synchronized (server.unopenedTablets) {
      synchronized (server.openingTablets) {
        synchronized (server.onlineTablets) {

          // checking if this exact tablet is in any of the sets
          // below is not a strong enough check
          // when splits and fix splits occurring

          Set<KeyExtent> unopenedOverlapping =
              KeyExtent.findOverlapping(extent, server.unopenedTablets);
          Set<KeyExtent> openingOverlapping =
              KeyExtent.findOverlapping(extent, server.openingTablets);
          Set<KeyExtent> onlineOverlapping =
              KeyExtent.findOverlapping(extent, server.getOnlineTablets());

          Set<KeyExtent> all = new HashSet<>();
          all.addAll(unopenedOverlapping);
          all.addAll(openingOverlapping);
          all.addAll(onlineOverlapping);

          if (!all.isEmpty()) {

            // ignore any tablets that have recently split, for error logging
            for (KeyExtent e2 : onlineOverlapping) {
              Tablet tablet = server.getOnlineTablet(e2);
              if (System.currentTimeMillis() - tablet.getSplitCreationTime()
                  < RECENTLY_SPLIT_MILLIES) {
                all.remove(e2);
              }
            }

            // ignore self, for error logging
            all.remove(extent);

            if (!all.isEmpty()) {
              log.error("Tablet {} overlaps previously assigned {} {} {}", extent,
                  unopenedOverlapping, openingOverlapping, onlineOverlapping + " " + all);
            }
            return;
          }

          server.unopenedTablets.add(extent);
        }
      }
    }

    TabletLogger.loading(extent, server.getTabletSession());

    final AssignmentHandler ah = new AssignmentHandler(server, extent);
    // final Runnable ah = new LoggingRunnable(log, );
    // Root tablet assignment must take place immediately

    if (extent.isRootTablet()) {
      Threads.createThread("Root Tablet Assignment", () -> {
        ah.run();
        if (server.getOnlineTablets().containsKey(extent)) {
          log.info("Root tablet loaded: {}", extent);
        } else {
          log.info("Root tablet failed to load");
        }
      }).start();
    } else {
      if (extent.isMeta()) {
        server.resourceManager.addMetaDataAssignment(extent, log, ah);
      } else {
        server.resourceManager.addAssignment(extent, log, ah);
      }
    }
  }

  @Override
  public void unloadTablet(TInfo tinfo, TCredentials credentials, String lock, TKeyExtent textent,
      TUnloadTabletGoal goal, long requestTime) {
    try {
      checkPermission(credentials, lock, "unloadTablet");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to unload a tablet", e);
      throw new RuntimeException(e);
    }

    KeyExtent extent = KeyExtent.fromThrift(textent);

    server.resourceManager.addMigration(extent,
        new UnloadTabletHandler(server, extent, goal, requestTime));
  }

  @Override
  public void flush(TInfo tinfo, TCredentials credentials, String lock, String tableId,
      ByteBuffer startRow, ByteBuffer endRow) {
    try {
      checkPermission(credentials, lock, "flush");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to flush a table", e);
      throw new RuntimeException(e);
    }

    ArrayList<Tablet> tabletsToFlush = new ArrayList<>();

    KeyExtent ke = new KeyExtent(TableId.of(tableId), ByteBufferUtil.toText(endRow),
        ByteBufferUtil.toText(startRow));

    for (Tablet tablet : server.getOnlineTablets().values()) {
      if (ke.overlaps(tablet.getExtent())) {
        tabletsToFlush.add(tablet);
      }
    }

    Long flushID = null;

    for (Tablet tablet : tabletsToFlush) {
      if (flushID == null) {
        // read the flush id once from zookeeper instead of reading
        // it for each tablet
        try {
          flushID = tablet.getFlushID();
        } catch (NoNodeException e) {
          // table was probably deleted
          log.info("Asked to flush table that has no flush id {} {}", ke, e.getMessage());
          return;
        }
      }
      tablet.flush(flushID);
    }
  }

  @Override
  public void flushTablet(TInfo tinfo, TCredentials credentials, String lock, TKeyExtent textent) {
    try {
      checkPermission(credentials, lock, "flushTablet");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to flush a tablet", e);
      throw new RuntimeException(e);
    }

    Tablet tablet = server.getOnlineTablet(KeyExtent.fromThrift(textent));
    if (tablet != null) {
      log.info("Flushing {}", tablet.getExtent());
      try {
        tablet.flush(tablet.getFlushID());
      } catch (NoNodeException nne) {
        log.info("Asked to flush tablet that has no flush id {} {}", KeyExtent.fromThrift(textent),
            nne.getMessage());
      }
    }
  }

  @Override
  public void halt(TInfo tinfo, TCredentials credentials, String lock)
      throws ThriftSecurityException {

    checkPermission(credentials, lock, "halt");

    Halt.halt(0, () -> {
      log.info("Master requested tablet server halt");
      server.gcLogger.logGCInfo(server.getConfiguration());
      server.requestStop();
      try {
        server.getLock().unlock();
      } catch (Exception e) {
        log.error("Caught exception unlocking TabletServer lock", e);
      }
    });
  }

  @Override
  public void fastHalt(TInfo info, TCredentials credentials, String lock) {
    try {
      halt(info, credentials, lock);
    } catch (Exception e) {
      log.warn("Error halting", e);
    }
  }

  @Override
  public TabletStats getHistoricalStats(TInfo tinfo, TCredentials credentials) {
    return server.statsKeeper.getTabletStats();
  }

  @Override
  public List<ActiveScan> getActiveScans(TInfo tinfo, TCredentials credentials)
      throws ThriftSecurityException, TException {
    try {
      checkPermission(credentials, null, "getScans");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to get active scans", e);
      throw e;
    }

    return server.sessionManager.getActiveScans();
  }

  @Override
  public void chop(TInfo tinfo, TCredentials credentials, String lock, TKeyExtent textent) {
    try {
      checkPermission(credentials, lock, "chop");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to chop extent", e);
      throw new RuntimeException(e);
    }

    KeyExtent ke = KeyExtent.fromThrift(textent);

    Tablet tablet = server.getOnlineTablet(ke);
    if (tablet != null) {
      tablet.chopFiles();
    }
  }

  @Override
  public void compact(TInfo tinfo, TCredentials credentials, String lock, String tableId,
      ByteBuffer startRow, ByteBuffer endRow) {
    try {
      checkPermission(credentials, lock, "compact");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to compact a table", e);
      throw new RuntimeException(e);
    }

    KeyExtent ke = new KeyExtent(TableId.of(tableId), ByteBufferUtil.toText(endRow),
        ByteBufferUtil.toText(startRow));

    Pair<Long,CompactionConfig> compactionInfo = null;

    for (Tablet tablet : server.getOnlineTablets().values()) {
      if (ke.overlaps(tablet.getExtent())) {
        // all for the same table id, so only need to read
        // compaction id once
        if (compactionInfo == null) {
          try {
            compactionInfo = tablet.getCompactionID();
          } catch (NoNodeException e) {
            log.info("Asked to compact table with no compaction id {} {}", ke, e.getMessage());
            return;
          }
        }
        tablet.compactAll(compactionInfo.getFirst(), compactionInfo.getSecond());
      }
    }
  }

  @Override
  public List<ActiveCompaction> getActiveCompactions(TInfo tinfo, TCredentials credentials)
      throws ThriftSecurityException, TException {
    try {
      checkPermission(credentials, null, "getActiveCompactions");
    } catch (ThriftSecurityException e) {
      log.error("Caller doesn't have permission to get active compactions", e);
      throw e;
    }

    List<CompactionInfo> compactions = Compactor.getRunningCompactions();
    List<ActiveCompaction> ret = new ArrayList<>(compactions.size());

    for (CompactionInfo compactionInfo : compactions) {
      ret.add(compactionInfo.toThrift());
    }

    return ret;
  }

  @Override
  public List<String> getActiveLogs(TInfo tinfo, TCredentials credentials) {
    String log = server.logger.getLogFile();
    // Might be null if there no active logger
    if (log == null) {
      return Collections.emptyList();
    }
    return Collections.singletonList(log);
  }

  @Override
  public void removeLogs(TInfo tinfo, TCredentials credentials, List<String> filenames) {
    log.warn("Garbage collector is attempting to remove logs through the tablet server");
    log.warn("This is probably because your file"
        + " Garbage Collector is an older version than your tablet servers.\n"
        + "Restart your file Garbage Collector.");
  }

  private TSummaries getSummaries(Future<SummaryCollection> future) throws TimeoutException {
    try {
      SummaryCollection sc =
          future.get(MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS, TimeUnit.MILLISECONDS);
      return sc.toThrift();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(e);
    } catch (ExecutionException e) {
      throw new RuntimeException(e);
    }
  }

  private TSummaries handleTimeout(long sessionId) {
    long timeout = server.getConfiguration().getTimeInMillis(Property.TSERV_CLIENT_TIMEOUT);
    server.sessionManager.removeIfNotAccessed(sessionId, timeout);
    return new TSummaries(false, sessionId, -1, -1, null);
  }

  private TSummaries startSummaryOperation(TCredentials credentials,
      Future<SummaryCollection> future) {
    try {
      return getSummaries(future);
    } catch (TimeoutException e) {
      long sid =
          server.sessionManager.createSession(new SummarySession(credentials, future), false);
      while (sid == 0) {
        server.sessionManager.removeSession(sid);
        sid = server.sessionManager.createSession(new SummarySession(credentials, future), false);
      }
      return handleTimeout(sid);
    }
  }

  @Override
  public TSummaries startGetSummaries(TInfo tinfo, TCredentials credentials,
      TSummaryRequest request)
      throws ThriftSecurityException, ThriftTableOperationException, TException {
    NamespaceId namespaceId;
    TableId tableId = TableId.of(request.getTableId());
    try {
      namespaceId = Tables.getNamespaceId(server.getContext(), tableId);
    } catch (TableNotFoundException e1) {
      throw new ThriftTableOperationException(tableId.canonical(), null, null,
          TableOperationExceptionType.NOTFOUND, null);
    }

    if (!security.canGetSummaries(credentials, tableId, namespaceId)) {
      throw new AccumuloSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED).asThriftException();
    }

    ExecutorService es = server.resourceManager.getSummaryPartitionExecutor();
    Future<SummaryCollection> future = new Gatherer(server.getContext(), request,
        context.getTableConfiguration(tableId), context.getCryptoService()).gather(es);

    return startSummaryOperation(credentials, future);
  }

  @Override
  public TSummaries startGetSummariesForPartition(TInfo tinfo, TCredentials credentials,
      TSummaryRequest request, int modulus, int remainder)
      throws ThriftSecurityException, TException {
    // do not expect users to call this directly, expect other tservers to call this method
    if (!security.canPerformSystemActions(credentials)) {
      throw new AccumuloSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED).asThriftException();
    }

    ExecutorService spe = server.resourceManager.getSummaryRemoteExecutor();
    TableConfiguration tableConfig =
        context.getTableConfiguration(TableId.of(request.getTableId()));
    Future<SummaryCollection> future =
        new Gatherer(server.getContext(), request, tableConfig, context.getCryptoService())
            .processPartition(spe, modulus, remainder);

    return startSummaryOperation(credentials, future);
  }

  @Override
  public TSummaries startGetSummariesFromFiles(TInfo tinfo, TCredentials credentials,
      TSummaryRequest request, Map<String,List<TRowRange>> files)
      throws ThriftSecurityException, TException {
    // do not expect users to call this directly, expect other tservers to call this method
    if (!security.canPerformSystemActions(credentials)) {
      throw new AccumuloSecurityException(credentials.getPrincipal(),
          SecurityErrorCode.PERMISSION_DENIED).asThriftException();
    }

    ExecutorService srp = server.resourceManager.getSummaryRetrievalExecutor();
    TableConfiguration tableCfg = context.getTableConfiguration(TableId.of(request.getTableId()));
    BlockCache summaryCache = server.resourceManager.getSummaryCache();
    BlockCache indexCache = server.resourceManager.getIndexCache();
    Cache<String,Long> fileLenCache = server.resourceManager.getFileLenCache();
    VolumeManager fs = context.getVolumeManager();
    FileSystemResolver volMgr = fs::getFileSystemByPath;
    Future<SummaryCollection> future =
        new Gatherer(server.getContext(), request, tableCfg, context.getCryptoService())
            .processFiles(volMgr, files, summaryCache, indexCache, fileLenCache, srp);

    return startSummaryOperation(credentials, future);
  }

  @Override
  public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
      throws NoSuchScanIDException, TException {
    SummarySession session = (SummarySession) server.sessionManager.getSession(sessionId);
    if (session == null) {
      throw new NoSuchScanIDException();
    }

    Future<SummaryCollection> future = session.getFuture();
    try {
      TSummaries tsums = getSummaries(future);
      server.sessionManager.removeSession(sessionId);
      return tsums;
    } catch (TimeoutException e) {
      return handleTimeout(sessionId);
    }
  }
}