text:
package com.github.ambry.replication;

import com.codahale.metrics.Counter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.github.ambry.clustermap.ClusterMap;
import com.github.ambry.clustermap.DataNodeId;
import com.github.ambry.clustermap.PartitionId;
import com.github.ambry.clustermap.ReplicaId;
import com.github.ambry.commons.BlobId;
import com.github.ambry.commons.ResponseHandler;
import com.github.ambry.commons.ServerErrorCode;
import com.github.ambry.config.ReplicationConfig;
import com.github.ambry.messageformat.DeleteMessageFormatInputStream;
import com.github.ambry.messageformat.MessageFormatException;
import com.github.ambry.messageformat.MessageFormatFlags;
import com.github.ambry.messageformat.MessageFormatInputStream;
import com.github.ambry.messageformat.MessageFormatWriteSet;
import com.github.ambry.messageformat.MessageSievingInputStream;
import com.github.ambry.messageformat.TtlUpdateMessageFormatInputStream;
import com.github.ambry.network.ChannelOutput;
import com.github.ambry.network.ConnectedChannel;
import com.github.ambry.network.ConnectionPool;
import com.github.ambry.notification.BlobReplicaSourceType;
import com.github.ambry.notification.NotificationSystem;
import com.github.ambry.notification.UpdateType;
import com.github.ambry.protocol.GetOption;
import com.github.ambry.protocol.GetRequest;
import com.github.ambry.protocol.GetResponse;
import com.github.ambry.protocol.PartitionRequestInfo;
import com.github.ambry.protocol.PartitionResponseInfo;
import com.github.ambry.protocol.ReplicaMetadataRequest;
import com.github.ambry.protocol.ReplicaMetadataRequestInfo;
import com.github.ambry.protocol.ReplicaMetadataResponse;
import com.github.ambry.protocol.ReplicaMetadataResponseInfo;
import com.github.ambry.store.FindToken;
import com.github.ambry.store.FindTokenFactory;
import com.github.ambry.store.MessageInfo;
import com.github.ambry.store.StoreErrorCodes;
import com.github.ambry.store.StoreException;
import com.github.ambry.store.StoreKey;
import com.github.ambry.store.StoreKeyConverter;
import com.github.ambry.store.Transformer;
import com.github.ambry.utils.ByteBufferInputStream;
import com.github.ambry.utils.SystemTime;
import com.github.ambry.utils.Time;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A replica thread is responsible for handling replication for a set of partitions assigned to it
 */
public class ReplicaThread implements Runnable {

  private final Map<DataNodeId, List<RemoteReplicaInfo>> replicasToReplicateGroupedByNode;
  private final Set<PartitionId> replicationDisabledPartitions = new HashSet<>();
  private final Set<PartitionId> unmodifiableReplicationDisabledPartitions =
      Collections.unmodifiableSet(replicationDisabledPartitions);
  private final Set<PartitionId> allReplicatedPartitions;
  private final CountDownLatch shutdownLatch = new CountDownLatch(1);
  private volatile boolean running;
  private final FindTokenFactory findTokenFactory;
  private final ClusterMap clusterMap;
  private final AtomicInteger correlationIdGenerator;
  private final DataNodeId dataNodeId;
  private final ConnectionPool connectionPool;
  private final ReplicationConfig replicationConfig;
  private final ReplicationMetrics replicationMetrics;
  private final String threadName;
  private final NotificationSystem notification;
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private final StoreKeyConverter storeKeyConverter;
  private final Transformer transformer;
  private final MetricRegistry metricRegistry;
  private final ResponseHandler responseHandler;
  private final boolean replicatingFromRemoteColo;
  private final boolean replicatingOverSsl;
  private final String datacenterName;
  private final long threadThrottleDurationMs;
  private final Time time;
  private final Counter throttleCount;
  private final Counter syncedBackOffCount;
  private final Counter idleCount;
  private final ReentrantLock lock = new ReentrantLock();
  private final Condition pauseCondition = lock.newCondition();

  private volatile boolean allDisabled = false;

  public ReplicaThread(String threadName, Map<DataNodeId, List<RemoteReplicaInfo>> replicasToReplicateGroupedByNode,
      FindTokenFactory findTokenFactory, ClusterMap clusterMap, AtomicInteger correlationIdGenerator,
      DataNodeId dataNodeId, ConnectionPool connectionPool, ReplicationConfig replicationConfig,
      ReplicationMetrics replicationMetrics, NotificationSystem notification, StoreKeyConverter storeKeyConverter,
      Transformer transformer, MetricRegistry metricRegistry, boolean replicatingOverSsl, String datacenterName,
      ResponseHandler responseHandler, Time time) {
    this.threadName = threadName;
    this.replicasToReplicateGroupedByNode = replicasToReplicateGroupedByNode;
    this.running = true;
    this.findTokenFactory = findTokenFactory;
    this.clusterMap = clusterMap;
    this.correlationIdGenerator = correlationIdGenerator;
    this.dataNodeId = dataNodeId;
    this.connectionPool = connectionPool;
    this.replicationConfig = replicationConfig;
    this.replicationMetrics = replicationMetrics;
    this.notification = notification;
    this.storeKeyConverter = storeKeyConverter;
    this.transformer = transformer;
    this.metricRegistry = metricRegistry;
    this.responseHandler = responseHandler;
    this.replicatingFromRemoteColo = !(dataNodeId.getDatacenterName().equals(datacenterName));
    this.replicatingOverSsl = replicatingOverSsl;
    this.datacenterName = datacenterName;
    this.time = time;
    Set<PartitionId> partitions = new HashSet<>();
    for (Map.Entry<DataNodeId, List<RemoteReplicaInfo>> entry : replicasToReplicateGroupedByNode.entrySet()) {
      for (RemoteReplicaInfo info : entry.getValue()) {
        partitions.add(info.getReplicaId().getPartitionId());
      }
    }
    allReplicatedPartitions = Collections.unmodifiableSet(partitions);
    if (replicatingFromRemoteColo) {
      threadThrottleDurationMs = replicationConfig.replicationInterReplicaThreadThrottleSleepDurationMs;
      syncedBackOffCount = replicationMetrics.interColoReplicaSyncedBackoffCount;
      idleCount = replicationMetrics.interColoReplicaThreadIdleCount;
      throttleCount = replicationMetrics.interColoReplicaThreadThrottleCount;
    } else {
      threadThrottleDurationMs = replicationConfig.replicationIntraReplicaThreadThrottleSleepDurationMs;
      syncedBackOffCount = replicationMetrics.intraColoReplicaSyncedBackoffCount;
      idleCount = replicationMetrics.intraColoReplicaThreadIdleCount;
      throttleCount = replicationMetrics.intraColoReplicaThreadThrottleCount;
    }
  }

  /**
   * Enables/disables replication on the given {@code ids}.
   * @param ids the {@link PartitionId}s to enable/disable it on.
   * @param enable whether to enable ({@code true}) or disable.
   */
  void controlReplicationForPartitions(Collection<PartitionId> ids, boolean enable) {
    lock.lock();
    try {
      for (PartitionId id : ids) {
        if (allReplicatedPartitions.contains(id)) {
          if (enable) {
            if (replicationDisabledPartitions.remove(id)) {
              allDisabled = false;
              pauseCondition.signal();
            }
          } else {
            replicationDisabledPartitions.add(id);
            allDisabled = allReplicatedPartitions.size() == replicationDisabledPartitions.size();
          }
          logger.info("Disable status of replication of {} from {} is {}. allDisabled for {} is {}", id, datacenterName,
              replicationDisabledPartitions.contains(id), getName(), allDisabled);
        }
      }
    } finally {
      lock.unlock();
    }
  }

  /**
   * @return {@link Set} of {@link PartitionId}s for which replication is disabled.
   */
  Set<PartitionId> getReplicationDisabledPartitions() {
    return unmodifiableReplicationDisabledPartitions;
  }

  String getName() {
    return threadName;
  }

  @Override
  public void run() {
    try {
      logger.trace("Starting replica thread on Local node: " + dataNodeId + " Thread name: " + threadName);
      List<List<RemoteReplicaInfo>> replicasToReplicate = new ArrayList<>(replicasToReplicateGroupedByNode.size());
      for (Map.Entry<DataNodeId, List<RemoteReplicaInfo>> replicasToReplicateEntry : replicasToReplicateGroupedByNode.entrySet()) {
        logger.info("Remote node: " + replicasToReplicateEntry.getKey() + " Thread name: " + threadName
            + " ReplicasToReplicate: " + replicasToReplicateEntry.getValue());
        replicasToReplicate.add(replicasToReplicateEntry.getValue());
      }
      logger.info("Begin iteration for thread " + threadName);
      while (running) {
        replicate(replicasToReplicate);
        lock.lock();
        try {
          if (running && allDisabled) {
            pauseCondition.await();
          }
        } catch (Exception e) {
          logger.error("Received interrupted exception during pause", e);
        } finally {
          lock.unlock();
        }
      }
    } finally {
      running = false;
      shutdownLatch.countDown();
    }
  }

  /**
   * Replicas from the given replicas
   * @param replicasToReplicate list of {@link RemoteReplicaInfo} by data node
   */
  public void replicate(List<List<RemoteReplicaInfo>> replicasToReplicate) {
    // shuffle the nodes
    Collections.shuffle(replicasToReplicate);
    boolean allCaughtUp = true;
    for (List<RemoteReplicaInfo> replicasToReplicatePerNode : replicasToReplicate) {
      if (!running) {
        break;
      }
      DataNodeId remoteNode = replicasToReplicatePerNode.get(0).getReplicaId().getDataNodeId();
      logger.trace("Remote node: {} Thread name: {} Remote replicas: {}", remoteNode, threadName,
          replicasToReplicatePerNode);
      Timer.Context context = null;
      Timer.Context portTypeBasedContext = null;
      if (replicatingFromRemoteColo) {
        context = replicationMetrics.interColoReplicationLatency.get(remoteNode.getDatacenterName()).time();
        if (replicatingOverSsl) {
          portTypeBasedContext =
              replicationMetrics.sslInterColoReplicationLatency.get(remoteNode.getDatacenterName()).time();
        } else {
          portTypeBasedContext =
              replicationMetrics.plainTextInterColoReplicationLatency.get(remoteNode.getDatacenterName()).time();
        }
      } else {
        context = replicationMetrics.intraColoReplicationLatency.time();
        if (replicatingOverSsl) {
          portTypeBasedContext = replicationMetrics.sslIntraColoReplicationLatency.time();
        } else {
          portTypeBasedContext = replicationMetrics.plainTextIntraColoReplicationLatency.time();
        }
      }
      ConnectedChannel connectedChannel = null;
      long checkoutConnectionTimeInMs = -1;
      long exchangeMetadataTimeInMs = -1;
      long fixMissingStoreKeysTimeInMs = -1;
      long replicationStartTimeInMs = SystemTime.getInstance().milliseconds();
      long startTimeInMs = replicationStartTimeInMs;

      List<RemoteReplicaInfo> activeReplicasPerNode = new ArrayList<RemoteReplicaInfo>();
      for (RemoteReplicaInfo remoteReplicaInfo : replicasToReplicatePerNode) {
        ReplicaId replicaId = remoteReplicaInfo.getReplicaId();
        boolean inBackoff = time.milliseconds() < remoteReplicaInfo.getReEnableReplicationTime();
        if (!replicationDisabledPartitions.contains(replicaId.getPartitionId()) && !replicaId.isDown() && !inBackoff) {
          activeReplicasPerNode.add(remoteReplicaInfo);
        }
      }
      if (activeReplicasPerNode.size() > 0) {
        allCaughtUp = false;
        try {
          connectedChannel =
              connectionPool.checkOutConnection(remoteNode.getHostname(), activeReplicasPerNode.get(0).getPort(),
                  replicationConfig.replicationConnectionPoolCheckoutTimeoutMs);
          checkoutConnectionTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
          startTimeInMs = SystemTime.getInstance().milliseconds();
          List<ExchangeMetadataResponse> exchangeMetadataResponseList =
              exchangeMetadata(connectedChannel, activeReplicasPerNode);
          exchangeMetadataTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;

          startTimeInMs = SystemTime.getInstance().milliseconds();
          fixMissingStoreKeys(connectedChannel, activeReplicasPerNode, exchangeMetadataResponseList);
          fixMissingStoreKeysTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
        } catch (Throwable e) {
          if (checkoutConnectionTimeInMs == -1) {
            // throwable happened in checkout connection phase
            checkoutConnectionTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
            responseHandler.onEvent(activeReplicasPerNode.get(0).getReplicaId(), e);
          } else if (exchangeMetadataTimeInMs == -1) {
            // throwable happened in exchange metadata phase
            exchangeMetadataTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
          } else if (fixMissingStoreKeysTimeInMs == -1) {
            // throwable happened in fix missing store phase
            fixMissingStoreKeysTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
          }
          logger.error("Error while talking to peer: Remote node: {}, Thread name: {}, Remote replicas: {}, Active "
                  + "remote replicas: {}, Checkout connection time: {}, Exchange metadata time: {}, Fix missing store key "
                  + "time {}", remoteNode, threadName, replicasToReplicatePerNode, activeReplicasPerNode,
              checkoutConnectionTimeInMs, exchangeMetadataTimeInMs, fixMissingStoreKeysTimeInMs, e);
          replicationMetrics.incrementReplicationErrors(replicatingOverSsl);
          if (connectedChannel != null) {
            connectionPool.destroyConnection(connectedChannel);
            connectedChannel = null;
          }
        } finally {
          long totalReplicationTime = SystemTime.getInstance().milliseconds() - replicationStartTimeInMs;
          replicationMetrics.updateTotalReplicationTime(totalReplicationTime, replicatingFromRemoteColo,
              replicatingOverSsl, datacenterName);
          if (connectedChannel != null) {
            connectionPool.checkInConnection(connectedChannel);
          }
          context.stop();
          portTypeBasedContext.stop();
        }
      }
    }
    long sleepDurationMs = 0;
    if (allCaughtUp && replicationConfig.replicationReplicaThreadIdleSleepDurationMs > 0) {
      sleepDurationMs = replicationConfig.replicationReplicaThreadIdleSleepDurationMs;
      idleCount.inc();
    } else if (threadThrottleDurationMs > 0) {
      sleepDurationMs = threadThrottleDurationMs;
      throttleCount.inc();
    }

    if (sleepDurationMs > 0) {
      try {
        long currentTime = time.milliseconds();
        time.sleep(sleepDurationMs);
        logger.trace("Replica thread: {} slept for {} ms", threadName, time.milliseconds() - currentTime);
      } catch (InterruptedException e) {
        logger.error("Received interrupted exception during throttling", e);
      }
    }
  }

  /**
   * Gets all the metadata about messages from the remote replicas since last token. Checks the messages with the local
   * store and finds all the messages that are missing. For the messages that are not missing, updates the delete
   * and ttl state.
   * @param connectedChannel The connected channel that represents a connection to the remote replica
   * @param replicasToReplicatePerNode The information about the replicas that is being replicated
   * @return - List of ExchangeMetadataResponse that contains the set of store keys that are missing from the local
   *           store and are present in the remote replicas and also the new token from the remote replicas
   * @throws IOException
   * @throws ReplicationException
   */
  List<ExchangeMetadataResponse> exchangeMetadata(ConnectedChannel connectedChannel,
      List<RemoteReplicaInfo> replicasToReplicatePerNode) throws IOException, ReplicationException {
    long exchangeMetadataStartTimeInMs = SystemTime.getInstance().milliseconds();
    List<ExchangeMetadataResponse> exchangeMetadataResponseList = new ArrayList<>();
    if (replicasToReplicatePerNode.size() > 0) {
      try {
        DataNodeId remoteNode = replicasToReplicatePerNode.get(0).getReplicaId().getDataNodeId();
        ReplicaMetadataResponse response =
            getReplicaMetadataResponse(replicasToReplicatePerNode, connectedChannel, remoteNode);
        long startTimeInMs = SystemTime.getInstance().milliseconds();

        Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = batchConvertReplicaMetadataResponseKeys(response);

        for (int i = 0; i < response.getReplicaMetadataResponseInfoList().size(); i++) {
          RemoteReplicaInfo remoteReplicaInfo = replicasToReplicatePerNode.get(i);
          ReplicaMetadataResponseInfo replicaMetadataResponseInfo =
              response.getReplicaMetadataResponseInfoList().get(i);
          responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
          if (replicaMetadataResponseInfo.getError() == ServerErrorCode.No_Error) {
            try {
              logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token from remote: {} Replica lag: {} ",
                  remoteNode, threadName, remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getFindToken(),
                  replicaMetadataResponseInfo.getRemoteReplicaLagInBytes());
              Set<StoreKey> missingStoreKeys =
                  getMissingStoreKeys(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
              processReplicaMetadataResponse(missingStoreKeys, replicaMetadataResponseInfo, remoteReplicaInfo,
                  remoteNode, remoteKeyToLocalKeyMap);
              ExchangeMetadataResponse exchangeMetadataResponse =
                  new ExchangeMetadataResponse(missingStoreKeys, replicaMetadataResponseInfo.getFindToken(),
                      replicaMetadataResponseInfo.getRemoteReplicaLagInBytes());
              exchangeMetadataResponseList.add(exchangeMetadataResponse);
            } catch (Exception e) {
              logger.error(
                  "Remote node: " + remoteNode + " Thread name: " + threadName + " Remote replica: " + remoteReplicaInfo
                      .getReplicaId(), e);
              replicationMetrics.updateLocalStoreError(remoteReplicaInfo.getReplicaId());
              responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), e);
              ExchangeMetadataResponse exchangeMetadataResponse =
                  new ExchangeMetadataResponse(ServerErrorCode.Unknown_Error);
              exchangeMetadataResponseList.add(exchangeMetadataResponse);
            }
          } else {
            replicationMetrics.updateMetadataRequestError(remoteReplicaInfo.getReplicaId());
            logger.error("Remote node: {} Thread name: {} Remote replica: {} Server error: {}", remoteNode, threadName,
                remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
            ExchangeMetadataResponse exchangeMetadataResponse =
                new ExchangeMetadataResponse(replicaMetadataResponseInfo.getError());
            exchangeMetadataResponseList.add(exchangeMetadataResponse);
          }
        }
        long processMetadataResponseTimeInMs = SystemTime.getInstance().milliseconds() - startTimeInMs;
        logger.trace("Remote node: {} Thread name: {} processMetadataResponseTime: {}", remoteNode, threadName,
            processMetadataResponseTimeInMs);
      } finally {
        long exchangeMetadataTime = SystemTime.getInstance().milliseconds() - exchangeMetadataStartTimeInMs;
        replicationMetrics.updateExchangeMetadataTime(exchangeMetadataTime, replicatingFromRemoteColo,
            replicatingOverSsl, datacenterName);
      }
    }
    return exchangeMetadataResponseList;
  }

  /**
   * Gets all the messages from the remote node for the missing keys and writes them to the local store
   * @param connectedChannel The connected channel that represents a connection to the remote replica
   * @param replicasToReplicatePerNode The information about the replicas that is being replicated
   * @param exchangeMetadataResponseList The missing keys in the local stores whose message needs to be retrieved
   *                                     from the remote stores
   * @throws IOException
   * @throws MessageFormatException
   * @throws ReplicationException
   */
  void fixMissingStoreKeys(ConnectedChannel connectedChannel, List<RemoteReplicaInfo> replicasToReplicatePerNode,
      List<ExchangeMetadataResponse> exchangeMetadataResponseList)
      throws IOException, MessageFormatException, ReplicationException {
    long fixMissingStoreKeysStartTimeInMs = SystemTime.getInstance().milliseconds();
    try {
      if (exchangeMetadataResponseList.size() != replicasToReplicatePerNode.size()
          || replicasToReplicatePerNode.size() == 0) {
        throw new IllegalArgumentException("ExchangeMetadataResponseList size " + exchangeMetadataResponseList.size()
            + " and replicasToReplicatePerNode size " + replicasToReplicatePerNode.size()
            + " should be the same and greater than zero");
      }
      DataNodeId remoteNode = replicasToReplicatePerNode.get(0).getReplicaId().getDataNodeId();
      GetResponse getResponse =
          getMessagesForMissingKeys(connectedChannel, exchangeMetadataResponseList, replicasToReplicatePerNode,
              remoteNode);
      writeMessagesToLocalStoreAndAdvanceTokens(exchangeMetadataResponseList, getResponse, replicasToReplicatePerNode,
          remoteNode);
    } finally {
      long fixMissingStoreKeysTime = SystemTime.getInstance().milliseconds() - fixMissingStoreKeysStartTimeInMs;
      replicationMetrics.updateFixMissingStoreKeysTime(fixMissingStoreKeysTime, replicatingFromRemoteColo,
          replicatingOverSsl, datacenterName);
    }
  }

  /**
   * Gets the replica metadata response for a list of remote replicas on a given remote data node
   * @param replicasToReplicatePerNode The list of remote replicas for a node
   * @param connectedChannel The connection channel to the node
   * @param remoteNode The remote node from which replication needs to happen
   * @return ReplicaMetadataResponse, the response from replica metadata request to remote node
   * @throws ReplicationException
   * @throws IOException
   */
  private ReplicaMetadataResponse getReplicaMetadataResponse(List<RemoteReplicaInfo> replicasToReplicatePerNode,
      ConnectedChannel connectedChannel, DataNodeId remoteNode) throws ReplicationException, IOException {
    long replicaMetadataRequestStartTime = SystemTime.getInstance().milliseconds();
    List<ReplicaMetadataRequestInfo> replicaMetadataRequestInfoList = new ArrayList<ReplicaMetadataRequestInfo>();
    for (RemoteReplicaInfo remoteReplicaInfo : replicasToReplicatePerNode) {
      ReplicaMetadataRequestInfo replicaMetadataRequestInfo =
          new ReplicaMetadataRequestInfo(remoteReplicaInfo.getReplicaId().getPartitionId(),
              remoteReplicaInfo.getToken(), dataNodeId.getHostname(),
              remoteReplicaInfo.getLocalReplicaId().getReplicaPath());
      replicaMetadataRequestInfoList.add(replicaMetadataRequestInfo);
      logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token going to be sent to remote: {} ",
          remoteNode, threadName, remoteReplicaInfo.getReplicaId(), remoteReplicaInfo.getToken());
    }

    try {
      ReplicaMetadataRequest request = new ReplicaMetadataRequest(correlationIdGenerator.incrementAndGet(),
          "replication-metadata-" + dataNodeId.getHostname(), replicaMetadataRequestInfoList,
          replicationConfig.replicationFetchSizeInBytes);
      connectedChannel.send(request);
      ChannelOutput channelOutput = connectedChannel.receive();
      ByteBufferInputStream byteBufferInputStream =
          new ByteBufferInputStream(channelOutput.getInputStream(), (int) channelOutput.getStreamSize());
      logger.trace("Remote node: {} Thread name: {} Remote replicas: {} ByteBuffer size after deserialization: {} ",
          remoteNode, threadName, replicasToReplicatePerNode, byteBufferInputStream.available());
      ReplicaMetadataResponse response =
          ReplicaMetadataResponse.readFrom(new DataInputStream(byteBufferInputStream), findTokenFactory, clusterMap);

      long metadataRequestTime = SystemTime.getInstance().milliseconds() - replicaMetadataRequestStartTime;
      replicationMetrics.updateMetadataRequestTime(metadataRequestTime, replicatingFromRemoteColo, replicatingOverSsl,
          datacenterName);

      if (response.getError() != ServerErrorCode.No_Error
          || response.getReplicaMetadataResponseInfoList().size() != replicasToReplicatePerNode.size()) {
        int replicaMetadataResponseInfoListSize = response.getReplicaMetadataResponseInfoList() == null ? 0
            : response.getReplicaMetadataResponseInfoList().size();
        logger.error("Remote node: " + remoteNode + " Thread name: " + threadName + " Remote replicas: "
            + replicasToReplicatePerNode + " Replica metadata response error: " + response.getError()
            + " ReplicaMetadataResponseInfoListSize: " + replicaMetadataResponseInfoListSize
            + " ReplicasToReplicatePerNodeSize: " + replicasToReplicatePerNode.size());
        throw new ReplicationException("Replica Metadata Response Error " + response.getError());
      }
      return response;
    } catch (IOException e) {
      responseHandler.onEvent(replicasToReplicatePerNode.get(0).getReplicaId(), e);
      throw e;
    }
  }

  /**
   * Gets the missing store keys by comparing the messages from the remote node
   * @param replicaMetadataResponseInfo The response that contains the messages from the remote node
   * @param remoteNode The remote node from which replication needs to happen
   * @param remoteReplicaInfo The remote replica that contains information about the remote replica id
   * @return List of store keys that are missing from the local store
   * @throws StoreException if store error (usually IOError) occurs when getting missing keys.
   */
  private Set<StoreKey> getMissingStoreKeys(ReplicaMetadataResponseInfo replicaMetadataResponseInfo,
      DataNodeId remoteNode, RemoteReplicaInfo remoteReplicaInfo) throws StoreException {
    long startTime = SystemTime.getInstance().milliseconds();
    List<MessageInfo> messageInfoList = replicaMetadataResponseInfo.getMessageInfoList();
    Map<StoreKey, StoreKey> remoteToConvertedNonNull = new HashMap<>();

    for (MessageInfo messageInfo : messageInfoList) {
      StoreKey storeKey = messageInfo.getStoreKey();
      logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key from remote: {}", remoteNode, threadName,
          remoteReplicaInfo.getReplicaId(), storeKey);
      StoreKey convertedKey = storeKeyConverter.getConverted(storeKey);
      if (convertedKey != null) {
        remoteToConvertedNonNull.put(storeKey, convertedKey);
      }
    }

    Set<StoreKey> convertedMissingStoreKeys =
        remoteReplicaInfo.getLocalStore().findMissingKeys(new ArrayList<>(remoteToConvertedNonNull.values()));
    Set<StoreKey> missingRemoteStoreKeys = new HashSet<>();
    remoteToConvertedNonNull.forEach((remoteKey, convertedKey) -> {
      if (convertedMissingStoreKeys.contains(convertedKey)) {
        logger.trace(
            "Remote node: {} Thread name: {} Remote replica: {} Key missing id (converted): {} Key missing id (remote): {}",
            remoteNode, threadName, remoteReplicaInfo.getReplicaId(), convertedKey, remoteKey);
        missingRemoteStoreKeys.add(remoteKey);
      }
    });

    replicationMetrics.updateCheckMissingKeysTime(SystemTime.getInstance().milliseconds() - startTime,
        replicatingFromRemoteColo, datacenterName);
    return missingRemoteStoreKeys;
  }

  /**
   * Takes the missing keys and the message list from the remote store and identifies messages that are deleted
   * on the remote store and updates them locally. Also, if the message that is missing is deleted in the remote
   * store, we remove the message from the list of missing keys
   * @param missingRemoteStoreKeys The list of keys missing from the local store
   * @param replicaMetadataResponseInfo The replica metadata response from the remote store
   * @param remoteReplicaInfo The remote replica that is being replicated from
   * @param remoteNode The remote node from which replication needs to happen
   * @param remoteKeyToLocalKeyMap map mapping remote keys to local key equivalents
   * @throws IOException
   * @throws StoreException
   * @throws MessageFormatException
   */
  private void processReplicaMetadataResponse(Set<StoreKey> missingRemoteStoreKeys,
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo, RemoteReplicaInfo remoteReplicaInfo,
      DataNodeId remoteNode, Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap)
      throws IOException, StoreException, MessageFormatException {
    long startTime = SystemTime.getInstance().milliseconds();
    List<MessageInfo> messageInfoList = replicaMetadataResponseInfo.getMessageInfoList();
    for (MessageInfo messageInfo : messageInfoList) {
      BlobId blobId = (BlobId) messageInfo.getStoreKey();
      if (remoteReplicaInfo.getLocalReplicaId().getPartitionId().compareTo(blobId.getPartition()) != 0) {
        throw new IllegalStateException(
            "Blob id is not in the expected partition Actual partition " + blobId.getPartition()
                + " Expected partition " + remoteReplicaInfo.getLocalReplicaId().getPartitionId());
      }
      BlobId localKey = (BlobId) remoteKeyToLocalKeyMap.get(messageInfo.getStoreKey());
      if (localKey == null) {
        missingRemoteStoreKeys.remove(messageInfo.getStoreKey());
        logger.trace("Remote node: {} Thread name: {} Remote replica: {} Remote key deprecated locally: {}", remoteNode,
            threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey());
      } else if (!missingRemoteStoreKeys.contains(messageInfo.getStoreKey())) {
        // the key is present in the local store. Mark it for deletion if it is deleted in the remote store and not
        // deleted yet locally
        boolean deletedLocally = remoteReplicaInfo.getLocalStore().isKeyDeleted(localKey);
        if (messageInfo.isDeleted() && !deletedLocally) {
          MessageFormatInputStream deleteStream =
              new DeleteMessageFormatInputStream(localKey, localKey.getAccountId(), localKey.getContainerId(),
                  messageInfo.getOperationTimeMs());
          MessageInfo info = new MessageInfo(localKey, deleteStream.getSize(), true, messageInfo.isTtlUpdated(),
              localKey.getAccountId(), localKey.getContainerId(), messageInfo.getOperationTimeMs());
          ArrayList<MessageInfo> infoList = new ArrayList<MessageInfo>();
          infoList.add(info);
          MessageFormatWriteSet writeset = new MessageFormatWriteSet(deleteStream, infoList, false);
          try {
            remoteReplicaInfo.getLocalStore().delete(writeset);
            logger.trace(
                "Remote node: {} Thread name: {} Remote replica: {} Key deleted. mark for deletion id: {} Local Key: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey(), localKey);
          } catch (StoreException e) {
            // The blob may get deleted between the time the above check is done and the delete is
            // attempted. For example, this can happen if the key gets deleted in the context of another replica
            // thread. This is more likely when replication is already caught up - when similar set of
            // messages are received from different replicas around the same time.
            if (e.getErrorCode() == StoreErrorCodes.ID_Deleted) {
              logger.trace(
                  "Remote node: {} Thread name: {} Remote replica: {} Remote Key already deleted: {} Local Key: {}",
                  remoteNode, threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey(), localKey);
            } else {
              throw e;
            }
          }
          // A Repair event for Delete signifies that a Delete message was received from the remote and it is fired
          // as long as the Delete is guaranteed to have taken effect locally.
          if (notification != null) {
            notification.onBlobReplicaDeleted(dataNodeId.getHostname(), dataNodeId.getPort(), localKey.getID(),
                BlobReplicaSourceType.REPAIRED);
          }
        } else if (!deletedLocally && !messageInfo.isDeleted() && messageInfo.isTtlUpdated()) {
          MessageInfo infoWithLocalKey =
              new MessageInfo(localKey, messageInfo.getSize(), false, true, messageInfo.getExpirationTimeInMs(),
                  messageInfo.getCrc(), localKey.getAccountId(), localKey.getContainerId(),
                  messageInfo.getOperationTimeMs());
          applyTtlUpdate(infoWithLocalKey, remoteReplicaInfo);
        }
      } else {
        if (messageInfo.isDeleted()) {
          // if the key is not present locally and if the remote replica has the message in deleted state,
          // it is not considered missing locally.
          missingRemoteStoreKeys.remove(messageInfo.getStoreKey());
          logger.trace(
              "Remote node: {} Thread name: {} Remote replica: {} Key in deleted state remotely: {} Local key: {}",
              remoteNode, threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey(), localKey);
          // A Repair event for Delete signifies that a Delete message was received from the remote and it is fired
          // as long as the Delete is guaranteed to have taken effect locally.
          if (notification != null) {
            notification.onBlobReplicaDeleted(dataNodeId.getHostname(), dataNodeId.getPort(), localKey.getID(),
                BlobReplicaSourceType.REPAIRED);
          }
        } else if (messageInfo.isExpired()) {
          // if the key is not present locally and if the remote replica has the key as expired,
          // it is not considered missing locally.
          missingRemoteStoreKeys.remove(messageInfo.getStoreKey());
          logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key in expired state remotely {}",
              remoteNode, threadName, remoteReplicaInfo.getReplicaId(), localKey);
        }
      }
    }
    if (replicatingFromRemoteColo) {
      replicationMetrics.interColoProcessMetadataResponseTime.get(datacenterName)
          .update(SystemTime.getInstance().milliseconds() - startTime);
    } else {
      replicationMetrics.intraColoProcessMetadataResponseTime.update(
          SystemTime.getInstance().milliseconds() - startTime);
    }
  }

  /**
   * Batch converts all keys in the {@link ReplicaMetadataResponse} response.
   * Intention is that conversion is done all at once so that followup calls to
   * {@link StoreKeyConverter#getConverted(StoreKey)} will work
   * @param response the {@link ReplicaMetadataResponse} whose keys will be converted
   * @return the map from the {@link StoreKeyConverter#convert(Collection)} call
   * @throws IOException thrown if {@link StoreKeyConverter#convert(Collection)} fails
   */
  private Map<StoreKey, StoreKey> batchConvertReplicaMetadataResponseKeys(ReplicaMetadataResponse response)
      throws IOException {
    try {
      List<StoreKey> storeKeysToConvert = new ArrayList<>();
      for (ReplicaMetadataResponseInfo replicaMetadataResponseInfo : response.getReplicaMetadataResponseInfoList()) {
        if ((replicaMetadataResponseInfo.getError() == ServerErrorCode.No_Error) && (
            replicaMetadataResponseInfo.getMessageInfoList() != null)) {
          for (MessageInfo messageInfo : replicaMetadataResponseInfo.getMessageInfoList()) {
            storeKeysToConvert.add(messageInfo.getStoreKey());
          }
        }
      }
      storeKeyConverter.dropCache();
      return storeKeyConverter.convert(storeKeysToConvert);
    } catch (Exception e) {
      throw new IOException("Problem with store key conversion", e);
    }
  }

  /**
   * Gets the messages for the keys that are missing from the local store by issuing a {@link GetRequest} to the remote
   * node, if there are any missing keys. If there are no missing keys to be fetched, then no request is issued and a
   * null response is returned.
   * @param connectedChannel The connection channel to the remote node
   * @param exchangeMetadataResponseList The list of metadata response from the remote node
   * @param replicasToReplicatePerNode The list of remote replicas for the remote node
   * @param remoteNode The remote node from which replication needs to happen
   * @return The response that contains the missing messages; or null if no request was issued because there were no
   * keys missing.
   * @throws ReplicationException
   * @throws IOException
   */
  private GetResponse getMessagesForMissingKeys(ConnectedChannel connectedChannel,
      List<ExchangeMetadataResponse> exchangeMetadataResponseList, List<RemoteReplicaInfo> replicasToReplicatePerNode,
      DataNodeId remoteNode) throws ReplicationException, IOException {
    List<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
    for (int i = 0; i < exchangeMetadataResponseList.size(); i++) {
      ExchangeMetadataResponse exchangeMetadataResponse = exchangeMetadataResponseList.get(i);
      RemoteReplicaInfo remoteReplicaInfo = replicasToReplicatePerNode.get(i);
      if (exchangeMetadataResponse.serverErrorCode == ServerErrorCode.No_Error) {
        Set<StoreKey> missingStoreKeys = exchangeMetadataResponse.missingStoreKeys;
        if (missingStoreKeys.size() > 0) {
          ArrayList<BlobId> keysToFetch = new ArrayList<BlobId>();
          for (StoreKey storeKey : missingStoreKeys) {
            keysToFetch.add((BlobId) storeKey);
          }
          PartitionRequestInfo partitionRequestInfo =
              new PartitionRequestInfo(remoteReplicaInfo.getReplicaId().getPartitionId(), keysToFetch);
          partitionRequestInfoList.add(partitionRequestInfo);
        }
      }
    }
    GetResponse getResponse = null;
    if (!partitionRequestInfoList.isEmpty()) {
      GetRequest getRequest = new GetRequest(correlationIdGenerator.incrementAndGet(),
          GetRequest.Replication_Client_Id_Prefix + dataNodeId.getHostname(), MessageFormatFlags.All,
          partitionRequestInfoList, replicationConfig.replicationIncludeAll ? GetOption.Include_All : GetOption.None);
      long startTime = SystemTime.getInstance().milliseconds();
      try {
        connectedChannel.send(getRequest);
        ChannelOutput channelOutput = connectedChannel.receive();
        getResponse = GetResponse.readFrom(new DataInputStream(channelOutput.getInputStream()), clusterMap);
        long getRequestTime = SystemTime.getInstance().milliseconds() - startTime;
        replicationMetrics.updateGetRequestTime(getRequestTime, replicatingFromRemoteColo, replicatingOverSsl,
            datacenterName);
        if (getResponse.getError() != ServerErrorCode.No_Error) {
          logger.error("Remote node: " + remoteNode + " Thread name: " + threadName + " Remote replicas: "
              + replicasToReplicatePerNode + " GetResponse from replication: " + getResponse.getError());
          throw new ReplicationException(
              " Get Request returned error when trying to get missing keys " + getResponse.getError());
        }
      } catch (IOException e) {
        responseHandler.onEvent(replicasToReplicatePerNode.get(0).getReplicaId(), e);
        throw e;
      }
    }
    return getResponse;
  }

  /**
   * Writes the messages (if any) to the local stores from the remote stores for the missing keys, and advances tokens.
   * @param exchangeMetadataResponseList The list of metadata response from the remote node
   * @param getResponse The {@link GetResponse} that contains the missing messages. This may be null if there are no
   *                    missing messages to write as per the exchange metadata response. In that case this method will
   *                    simply advance the tokens for every store.
   * @param replicasToReplicatePerNode The list of remote replicas for the remote node
   * @param remoteNode The remote node from which replication needs to happen
   * @throws IOException
   * @throws MessageFormatException
   */
  private void writeMessagesToLocalStoreAndAdvanceTokens(List<ExchangeMetadataResponse> exchangeMetadataResponseList,
      GetResponse getResponse, List<RemoteReplicaInfo> replicasToReplicatePerNode, DataNodeId remoteNode)
      throws IOException, MessageFormatException {
    int partitionResponseInfoIndex = 0;
    long totalBytesFixed = 0;
    long totalBlobsFixed = 0;
    long startTime = SystemTime.getInstance().milliseconds();
    for (int i = 0; i < exchangeMetadataResponseList.size(); i++) {
      ExchangeMetadataResponse exchangeMetadataResponse = exchangeMetadataResponseList.get(i);
      RemoteReplicaInfo remoteReplicaInfo = replicasToReplicatePerNode.get(i);
      if (exchangeMetadataResponse.serverErrorCode == ServerErrorCode.No_Error) {
        if (remoteReplicaInfo.getToken().equals(exchangeMetadataResponse.remoteToken)) {
          remoteReplicaInfo.setReEnableReplicationTime(
              time.milliseconds() + replicationConfig.replicationSyncedReplicaBackoffDurationMs);
          syncedBackOffCount.inc();
        }
        if (exchangeMetadataResponse.missingStoreKeys.size() > 0) {
          PartitionResponseInfo partitionResponseInfo =
              getResponse.getPartitionResponseInfoList().get(partitionResponseInfoIndex);
          responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), partitionResponseInfo.getErrorCode());
          partitionResponseInfoIndex++;
          if (partitionResponseInfo.getPartition().compareTo(remoteReplicaInfo.getReplicaId().getPartitionId()) != 0) {
            throw new IllegalStateException(
                "The partition id from partitionResponseInfo " + partitionResponseInfo.getPartition()
                    + " and from remoteReplicaInfo " + remoteReplicaInfo.getReplicaId().getPartitionId()
                    + " are not the same");
          }
          if (partitionResponseInfo.getErrorCode() == ServerErrorCode.No_Error) {
            try {
              List<MessageInfo> messageInfoList = partitionResponseInfo.getMessageInfoList();
              logger.trace("Remote node: {} Thread name: {} Remote replica: {} Messages to fix: {} "
                      + "Partition: {} Local mount path: {}", remoteNode, threadName, remoteReplicaInfo.getReplicaId(),
                  exchangeMetadataResponse.missingStoreKeys, remoteReplicaInfo.getReplicaId().getPartitionId(),
                  remoteReplicaInfo.getLocalReplicaId().getMountPath());

              MessageFormatWriteSet writeset;
              MessageSievingInputStream validMessageDetectionInputStream =
                  new MessageSievingInputStream(getResponse.getInputStream(), messageInfoList,
                      Collections.singletonList(transformer), metricRegistry);
              if (validMessageDetectionInputStream.hasInvalidMessages()) {
                replicationMetrics.incrementInvalidMessageError(partitionResponseInfo.getPartition());
                logger.error("Out of " + (messageInfoList.size()) + " messages, " + (messageInfoList.size()
                    - validMessageDetectionInputStream.getValidMessageInfoList().size())
                    + " invalid messages were found in message stream from " + remoteReplicaInfo.getReplicaId());
              }
              messageInfoList = validMessageDetectionInputStream.getValidMessageInfoList();
              if (messageInfoList.size() == 0) {
                logger.debug(
                    "MessageInfoList is of size 0 as all messages are invalidated, deprecated, deleted or expired.");
              } else {
                writeset = new MessageFormatWriteSet(validMessageDetectionInputStream, messageInfoList, false);
                remoteReplicaInfo.getLocalStore().put(writeset);
              }

              for (MessageInfo messageInfo : messageInfoList) {
                totalBytesFixed += messageInfo.getSize();
                logger.trace("Remote node: {} Thread name: {} Remote replica: {} Message replicated: {} Partition: {} "
                        + "Local mount path: {} Message size: {}", remoteNode, threadName, remoteReplicaInfo.getReplicaId(),
                    messageInfo.getStoreKey(), remoteReplicaInfo.getReplicaId().getPartitionId(),
                    remoteReplicaInfo.getLocalReplicaId().getMountPath(), messageInfo.getSize());
                if (notification != null) {
                  notification.onBlobReplicaCreated(dataNodeId.getHostname(), dataNodeId.getPort(),
                      messageInfo.getStoreKey().getID(), BlobReplicaSourceType.REPAIRED);
                }
                if (messageInfo.isTtlUpdated()) {
                  applyTtlUpdate(messageInfo, remoteReplicaInfo);
                }
              }
              totalBlobsFixed += messageInfoList.size();
              remoteReplicaInfo.setToken(exchangeMetadataResponse.remoteToken);
              remoteReplicaInfo.setLocalLagFromRemoteInBytes(exchangeMetadataResponse.localLagFromRemoteInBytes);
              logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token after speaking to remote node: {}",
                  remoteNode, threadName, remoteReplicaInfo.getReplicaId(), exchangeMetadataResponse.remoteToken);
            } catch (StoreException e) {
              if (e.getErrorCode() != StoreErrorCodes.Already_Exist) {
                replicationMetrics.updateLocalStoreError(remoteReplicaInfo.getReplicaId());
                logger.error("Remote node: " + remoteNode + " Thread name: " + threadName + " Remote replica: "
                    + remoteReplicaInfo.getReplicaId(), e);
              }
            }
          } else if (partitionResponseInfo.getErrorCode() == ServerErrorCode.Blob_Deleted) {
            replicationMetrics.blobDeletedOnGetCount.inc();
            logger.trace("One of the blobs to GET is deleted: Remote node: {} Thread name: {} Remote replica: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId());
          } else if (partitionResponseInfo.getErrorCode() == ServerErrorCode.Blob_Authorization_Failure) {
            replicationMetrics.blobAuthorizationFailureCount.inc();
            logger.error(
                "One of the blobs authorization failed: Remote node: {} Thread name: {} Remote replica: {} Keys are: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId(), exchangeMetadataResponse.missingStoreKeys);
          } else {
            replicationMetrics.updateGetRequestError(remoteReplicaInfo.getReplicaId());
            logger.error("Remote node: {} Thread name: {} Remote replica: {} Server error: {}", remoteNode, threadName,
                remoteReplicaInfo.getReplicaId(), partitionResponseInfo.getErrorCode());
          }
        } else {
          // There are no missing keys. We just advance the token
          remoteReplicaInfo.setToken(exchangeMetadataResponse.remoteToken);
          remoteReplicaInfo.setLocalLagFromRemoteInBytes(exchangeMetadataResponse.localLagFromRemoteInBytes);
          logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token after speaking to remote node: {}",
              remoteNode, threadName, remoteReplicaInfo.getReplicaId(), exchangeMetadataResponse.remoteToken);
        }
      }
    }
    long batchStoreWriteTime = SystemTime.getInstance().milliseconds() - startTime;
    replicationMetrics.updateBatchStoreWriteTime(batchStoreWriteTime, totalBytesFixed, totalBlobsFixed,
        replicatingFromRemoteColo, replicatingOverSsl, datacenterName);
  }

  /**
   * Applies a TTL update to the blob described by {@code messageInfo}.
   * @param messageInfo the {@link MessageInfo} that will be transformed into a TTL update
   * @param remoteReplicaInfo The remote replica that is being replicated from
   * @throws IOException
   * @throws MessageFormatException
   * @throws StoreException
   */
  private void applyTtlUpdate(MessageInfo messageInfo, RemoteReplicaInfo remoteReplicaInfo)
      throws IOException, MessageFormatException, StoreException {
    MessageFormatInputStream ttlUpdateStream =
        new TtlUpdateMessageFormatInputStream(messageInfo.getStoreKey(), messageInfo.getAccountId(),
            messageInfo.getContainerId(), messageInfo.getExpirationTimeInMs(), messageInfo.getOperationTimeMs());
    MessageInfo info = new MessageInfo(messageInfo.getStoreKey(), ttlUpdateStream.getSize(), false, true,
        messageInfo.getExpirationTimeInMs(), messageInfo.getAccountId(), messageInfo.getContainerId(),
        messageInfo.getOperationTimeMs());
    MessageFormatWriteSet writeSet = new MessageFormatWriteSet(ttlUpdateStream, Collections.singletonList(info), false);
    DataNodeId remoteNode = remoteReplicaInfo.getReplicaId().getDataNodeId();
    try {
      // NOTE: It is possible that the key in question may have expired and this TTL update is being applied after it
      // is deemed expired. The store will accept the op (BlobStore looks at whether the op was valid to do at the time
      // of the op, not current time) but if compaction is running at the same time and has decided to clean up the
      // record before this ttl update was applied (and this didn't find the key missing because compaction has not yet
      // committed), then we have a bad situation where only a TTL update exists in the store. This problem has to be
      // addressed. This can only happen if replication is far behind (for e.g due to a server being down for a long
      // time). Won't happen if a server is being recreated.
      remoteReplicaInfo.getLocalStore().updateTtl(writeSet);
      logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key ttl updated id: {}", remoteNode, threadName,
          remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey());
    } catch (StoreException e) {
      // The blob may be deleted or updated which is alright
      if (e.getErrorCode() == StoreErrorCodes.ID_Deleted || e.getErrorCode() == StoreErrorCodes.Already_Updated) {
        logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key already updated: {}", remoteNode,
            threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey());
      } else {
        throw e;
      }
    }
    // A Repair event for an update signifies that an update message was received from the remote and it is fired
    // as long as the update is guaranteed to have taken effect locally.
    if (notification != null) {
      notification.onBlobReplicaUpdated(dataNodeId.getHostname(), dataNodeId.getPort(),
          messageInfo.getStoreKey().getID(), BlobReplicaSourceType.REPAIRED, UpdateType.TTL_UPDATE, messageInfo);
    }
  }

  static class ExchangeMetadataResponse {
    final Set<StoreKey> missingStoreKeys;
    final FindToken remoteToken;
    final long localLagFromRemoteInBytes;
    final ServerErrorCode serverErrorCode;

    ExchangeMetadataResponse(Set<StoreKey> missingStoreKeys, FindToken remoteToken, long localLagFromRemoteInBytes) {
      this.missingStoreKeys = missingStoreKeys;
      this.remoteToken = remoteToken;
      this.localLagFromRemoteInBytes = localLagFromRemoteInBytes;
      this.serverErrorCode = ServerErrorCode.No_Error;
    }

    ExchangeMetadataResponse(ServerErrorCode errorCode) {
      missingStoreKeys = null;
      remoteToken = null;
      localLagFromRemoteInBytes = -1;
      this.serverErrorCode = errorCode;
    }
  }

  boolean isThreadUp() {
    return running;
  }

  void shutdown() throws InterruptedException {
    running = false;
    lock.lock();
    try {
      pauseCondition.signal();
    } finally {
      lock.unlock();
    }
    shutdownLatch.await();
  }
}


text:
import { Injectable } from '@angular/core';
import { Platform } from 'ionic-angular';

@Injectable()
export class MapsService {
	private platform: Platform;

	constructor(platform: Platform) {
		this.platform = platform;
	}

	public openMapsApp(location: any) {
		let q;
		if (this.platform.is('android')) {
			q = 'geo:' + location;
		} else {
			q = 'maps://maps.apple.com/?q=' + location;
		}
		window.location.href = q;
	}
}

redpajama_set_name: RedPajamaGithub


text:
    <meta http-equiv="Content-Type" content="text/html; charset=Windows-1252" />
    <meta name="vs_targetSchema" content="http://schemas.microsoft.com/intellisense/ie5" />
    <title>ConversionNotSupportedException Class</title>
    <xml>
    </xml>
    <link rel="stylesheet" type="text/css" href="MSDN.css" />

            Apache log4net™ SDK Documentation - Microsoft .NET Framework 4.0

ConversionNotSupportedException Class

 Exception base type for conversion errors.

For a list of all members of this type, see ConversionNotSupportedException Members.

        System.Object

   System.Exception
      System.ApplicationException
         log4net.Util.TypeConverters.ConversionNotSupportedException

        [Visual Basic]

Public Class ConversionNotSupportedException
    Inherits ApplicationException

        [C#]

public class ConversionNotSupportedException : ApplicationException

Thread Safety

Public static (Shared in Visual Basic) members of this type are
				safe for multithreaded operations. Instance members are not guaranteed to be
				thread-safe.

Remarks

 This type extends ApplicationException. It does not add any new functionality but does differentiate the type of exception being thrown.

Requirements

Namespace: log4net.Util.TypeConverters

Assembly: log4net (in log4net.dll)

See Also

ConversionNotSupportedException Members | log4net.Util.TypeConverters Namespace
<object type="application/x-oleobject" classid="clsid:1e2a7bd0-dab9-11d0-b93a-00c04fc99f9e" viewastext="true" style="display: none;"><param name="Keyword" value="ConversionNotSupportedException class, about ConversionNotSupportedException class"></param></object>

Copyright 2004-2013 The Apache Software Foundation.

Apache log4net, Apache and log4net are trademarks of The Apache Software Foundation.


text:
ACCEPTED

#### According to
Index Fungorum

#### Published in
in Crous & Braun, CBS Diversity Ser. (Utrecht) 1: 294 (2003)

#### Original name
Cercospora obtegens Syd. {?}

### Remarks
null


text:
import sys, os, re

# Process elasticsearch hits and return flights records
def process_search(results):
  records = []
  total = 0
  if results['hits'] and results['hits']['hits']:
    total = results['hits']['total']
    hits = results['hits']['hits']
    for hit in hits:
      record = hit['_source']
      records.append(record)
  return records, total

# Calculate offsets for fetching lists of flights from MongoDB
def get_navigation_offsets(offset1, offset2, increment):
  offsets = {}
  offsets['Next'] = {'top_offset': offset2 + increment, 'bottom_offset':
  offset1 + increment}
  offsets['Previous'] = {'top_offset': max(offset2 - increment, 0),
 'bottom_offset': max(offset1 - increment, 0)} # Don't go < 0
  return offsets

# Strip the existing start and end parameters from the query string
def strip_place(url):
  try:
    p = re.match('(.+)\?start=.+&end=.+', url).group(1)
  except AttributeError, e:
    return url
  return p


text:
var context = require.context('./spec/client', true, /.+\Spec\.js?$/);

context.keys().forEach(context);
module.exports = context;


text:
<?xml version="1.0" encoding="utf-8"?>
<!-- Copyright (C) 2013 The Android Open Source Project

     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.
-->

<resources>
    <style name="Theme.WallpaperCropper" parent="@android:style/Theme.Material.Light">
        <item name="android:actionBarStyle">@style/WallpaperCropperActionBar</item>
        <item name="android:windowFullscreen">true</item>
        <item name="android:windowActionBarOverlay">true</item>
    </style>

    <style name="WallpaperCropperActionBar" parent="@android:style/Widget.Material.ActionBar">
        <item name="android:displayOptions">showCustom</item>
        <item name="android:background">#88000000</item>
    </style>
</resources>


text:
using Intense.Presentation;

namespace Intense.TestApp
{
    public sealed partial class NavigationStructure : NavigationItem
    {
        public NavigationStructure()
        {
            this.InitializeComponent();
        }
    }
}


text:
namespace Torken.CQRS.Core.Exceptions
{
    using System;

    public class QueryHandlerNotFound : Exception
    {
    }
}


text:
using System.Reflection.Metadata.Ecma335;

namespace Internal.TypeSystem.Ecma
{
    // Functionality related to determinstic ordering of types
    partial class EcmaType
    {
        protected internal override int ClassCode => 1340416537;

        protected internal override int CompareToImpl(TypeDesc other, TypeSystemComparer comparer)
        {
            var otherType = (EcmaType)other;
            int result = _module.MetadataReader.GetToken(_handle) - otherType._module.MetadataReader.GetToken(otherType._handle);
            if (result != 0)
                return result;

            return _module.CompareTo(otherType._module);
        }
    }
}


text:
package io.dummymaker.annotation.simple.string;

import io.dummymaker.annotation.core.PrimeGen;
import io.dummymaker.generator.simple.string.HouseGenerator;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * @author GoodforGod
 * @see HouseGenerator
 * @since 12.10.2019
 */
@PrimeGen(HouseGenerator.class)
@Retention(value = RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
public @interface GenHouse {

}


text:
	@if (Menu::exists('leftMenu') && count(Menu::getMenu('leftMenu')->items) > 0)

			@each('layouts.menus.utopian.item', Menu::getMenu('leftMenu')->items, 'item')

	@endif
	@if (Menu::exists('rightMenu') && count(Menu::getMenu('rightMenu')->items) > 0)

				@each('layouts.menus.utopian.item', Menu::getMenu('rightMenu')->items, 'item')

	@endif


text:
MathSwift
=========
MathSwift is an iOS Dynamic Framework written in Swift which supports complex mathmatical operations. Currently it mainly support matrix and some advanced linear  algrbra operations such as singular value decomposition. It may also serve as a demostration of good practice for developing iOS/OSX dynamic frameworks in pure Swift, including a complete set of unit tests.
# Installation
MathSwift supports installation using [Carthage](https://github.com/Carthage/Carthage). You can integrate MathSwift with your project by adding the following line in your [Cartfile](https://github.com/Carthage/Carthage/blob/master/Documentation/Artifacts.md#cartfile):
```
github "huangenyan/MathSwift"
```
To use MathSwift in your code, just remember to import it in your source code files:
```Swift
import MathSwift
```
# Usage
The full documentation of MathSwift isn't available yet. However, each public method is well commented in the code so you may refer to it. Here I just show some examples.
## Create a matrix
There are several ways for initializing a matrix:
```Swift
// Create with dimension
let m1 = Matrix(rows:3, columns:3)

// Create with 2D-array
let m2 = Matrix(elements:[[11,12,13],[21,22,23],[31,32,33]])
```
## Access elements
You can access the element(s) in the matrix by using the subscript syntax with the format *m[row(s), column(s)]*:
```Swift
var m = Matrix(elements:[[11,12,13],[21,22,23],[31,32,33]])

// Index is an integer. subscript syntax will always return a Matrix so we need to convert it to double manually.
// Get
let a11: Double = m[0,0].toDouble()!
// Set
m[0,0] = a11.toMatrix()
m[0,0] = 0.0 // Double literal will be automatically converted to Matrix

// Index is an array.
// Get
let firstCol = m[[0,1,2],0] // == [[11],[21],[31]]
let firstLastCol = m[[0,1,2],[0,2]] // == [[11,13],[21,23],[31,33]]
// Set
m[[0,1,2],0] = firstCol

// Index is a range.
// Get
let firstTwoCol = m[[0,1,2],0..<2]
let upperLeft4 = m[0...1,0...1]
// Set
m[0...1,0...1] = upperLeft4

// Index is `ALL`
// Get
let firstRow = m[0,ALL] // equals to m[0,[0,1,2]]
// Set
m[0,ALL] = firstRow
```
## Operations on matrix
I hope the code below will explain everything about operators :]. Please note this is not Swift code but just some intuitive expressions, which is similar to the matrix expression in Matlab.
```
// Addition and subtraction
[1 2; 3 4] + [1 1; 1 1] = [2 3; 4 5]
[1 2; 3 4] - [1 1; 1 1] = [0 1; 2 3]
[1 2; 3 4] + 1 = [2 3; 4 5]
1 + [1 2; 3 4] = [2 3; 4 5]
[1 2; 3 4] - 1 = [0 1; 2 3]
-[1 2; 3 4] = [-1 -2; -3 -4]
1 - [1 2; 3 4] = [0 -1; -2 -3]

// Element-wise multiplication/division
[1 2; 3 4] *~ [2 2; 2 2] = [2 4; 6 8]
[2 4; 6 8] /~ [2 2; 2 2] = [2 4; 6 8]

// Scalar multiplication/division
[1 2; 3 4] * 2 = [2 4; 6 8]
2 * [1 2; 3 4] = [2 4; 6 8]
[2 4; 6 8] / 2 = [1 2; 3 4]
12 / [1 2; 3 4] = [12 6; 4 3]

// Element-wise exponentiation
[1 2; 3 4] ^~ 2 = [1 4; 9 16]

// Matrix exponentiation
[1 2; 3 4] ^ 2 = [1 2; 3 4] * [1 2; 3 4] = [7 10; 15 22]

// Production of two matrices
[1 2; 3 4] * [1 2; 3 4] = [7 10; 15 22]

// Horizontal concantenation
[1 2; 3 4] +++ [5; 6] = [1 2 5; 3 4 6]

// Vertical concantenation
[1 2; 3 4] --- [5 6] = [1 2; 3 4; 5 6]

```
The transpose, inverse and some other properties of a matrix is available as well. For example, you can get the transpose of a matrix by `m.transpose`.


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<title>RoundList (JADE v4.3.1 API)</title>
<meta name="date" content="2013-12-04">
<link rel="stylesheet" type="text/css" href="../../../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

jade.util.leap

Class RoundList

java.lang.Object

jade.util.leap.RoundList

All Implemented Interfaces:

java.io.Serializable

public class RoundList
extends java.lang.Object
implements Serializable

Implementation of a RoundList with get/insert methods relative
 to the current element

Version:

$Date: 2005-11-03 11:40:16 +0100 (gio, 03 nov 2005) $ $Revision: 5810 $

Author:

Fabio Bellifemine - TILab

See Also:

Serialized Form

Constructor Summary

Constructors

Constructor and Description

RoundList()

Default constructor.

Method Summary

Methods

Modifier and Type
Method and Description

boolean
add(java.lang.Object element)

Inserts the element before the current element.

boolean
contains(java.lang.Object element)

Returns true if this list contains the specified element.

java.lang.Object
get()

Returns the current element in the list and updates the pointer
 such that the current becomes the
 next element in the list.

Iterator
iterator()

Returns an Iterator over the elements in this list.

static void
main(java.lang.String[] args)

Just for Debugging this implementation.

boolean
remove(java.lang.Object element)

Removes the first occurrence of the specified element in this list
  and updates the pointer to the current element.

int
size()

Returns the number of elements in this list.

java.lang.Object[]
toArray()

java.lang.String
toString()

Returns a string representation of this collection.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

RoundList

public RoundList()

Default constructor.

Method Detail

add

public boolean add(java.lang.Object element)

Inserts the element before the current element.
 If the list was empty, the inserted element becomes also the current element.
  Note that this implementation uses a LinkedList
 and therefore it is not synchronized.

Parameters:

element - the element to insert

Returns:

true (as per the general contract of Collection.add).

get

public java.lang.Object get()
                     throws java.util.NoSuchElementException

Returns the current element in the list and updates the pointer
 such that the current becomes the
 next element in the list.

 Notice that if the list contains just 1 element each call to this method will return
 the same element.

 Take care in avoiding infinite loops in calling this method. It
 must be called no more than size() times

Throws:

java.util.NoSuchElementException - if the list is empty

remove

public boolean remove(java.lang.Object element)

Removes the first occurrence of the specified element in this list
  and updates the pointer to the current element.
  If the list does not contain the element, it is unchanged.
  More formally, removes the element with the lowest index i such that
 (element==null ? get(i)==null : element.equals(get(i)))
 (if such an element exists).

Parameters:

element - the element to be removed from this list, if present.

Returns:

true if the list contained the specified element.

contains

public boolean contains(java.lang.Object element)

Returns true if this list contains the specified element.
 More formally, returns true if and only if this list contains at least
 one element e such that (element==null ? e==null : element.equals(e)).

Parameters:

element - whose presence in this list is to be tested.

Returns:

true if this list contains the specified element.

size

public int size()

Returns the number of elements in this list.

Returns:

the number of elements in this list.

iterator

public Iterator iterator()

Returns an Iterator over the elements in this list.

Returns:

an Iterator over the elements in this list.

toArray

public java.lang.Object[] toArray()

toString

public java.lang.String toString()

Returns a string representation of this collection.
 The string representation consists of a list of the collection's elements
 in the order they are returned by its get() method,
 enclosed in square brackets ("[]").
 Adjacent elements are separated by the characters ", " (comma and space).
 Elements are converted to strings as by String.valueOf(Object).

Overrides:

toString in class java.lang.Object

Returns:

a String representation of this list

main

public static void main(java.lang.String[] args)

Just for Debugging this implementation.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

These are the official JADE API. For these API backward compatibility is guaranteed accross JADE versions


text:
import { CommonModule } from "@angular/common";
import { NgModule } from "@angular/core";
import { FormsModule } from "@angular/forms";
import { RouterModule, Routes } from "@angular/router";

import { AdditionalGoalsComponent } from "./additional-goals.component";
import { DashboardComponent } from "./dashboard.component";
import { GlobalGoalsComponent } from "./global-goals.component";
import { GoalsPriorityComponent } from "./goals-priority.component";
import { SponsorsComponent } from "./sponsors.component";

import { CanDeactivateGuard } from "../core/core-router.module";
import { NgChartModule } from "../ng-chart/ng-chart.module";
import { ResourcePoolModule } from "../resource-pool/resource-pool.module";

// Routes
export const globalGoalsFundRoutes: Routes = [
    { path: "", component: GlobalGoalsComponent, data: { title: "Global Goals" }, canDeactivate: [CanDeactivateGuard] },
    { path: "app/global-goals", component: GlobalGoalsComponent, data: { title: "Global Goals" }, canDeactivate: [CanDeactivateGuard] },
    { path: "app/additional-goals", component: AdditionalGoalsComponent, data: { title: "Additional Goals" }, canDeactivate: [CanDeactivateGuard] },
    { path: "app/goals-priority", component: GoalsPriorityComponent, data: { title: "Goals Priority" }, canDeactivate: [CanDeactivateGuard] },
    { path: "app/sponsors", component: SponsorsComponent, data: { title: "Sponsors" } },
    { path: "app/dashboard", component: DashboardComponent, data: { title: "Dashboard" }, canDeactivate: [CanDeactivateGuard] },

    // /* Home alternatives */
    { path: "app/global-goals", redirectTo: "", pathMatch: "full" },
];

@NgModule({
    declarations: [
        AdditionalGoalsComponent,
        DashboardComponent,
        GlobalGoalsComponent,
        GoalsPriorityComponent,
        SponsorsComponent,
    ],
    exports: [
        AdditionalGoalsComponent,
        DashboardComponent,
        GlobalGoalsComponent,
        GoalsPriorityComponent,
        SponsorsComponent,
    ],
    imports: [
        CommonModule,
        FormsModule,
        RouterModule,

        NgChartModule,
        ResourcePoolModule
    ]
})
export class GlobalGoalsFundModule { }


text:
ACCEPTED

#### According to
Index Fungorum

#### Published in
Proc. R. Soc. N. S. W. 72: 211 (1947)

#### Original name
Microthyriacites fimbriatus Cookson

### Remarks
null


text:
from flask_restful import Api, Resource

from modules.hello.api import load_api as hello_api
from settings import API_VERSION

class API(Resource):
    def get(self):
        return 'Hello API', 200, {'API-Version': API_VERSION}

def load_api(app):
    Api(app).add_resource(API, '/api')
    hello_api(app)


text:
#include "dcmtk/config/osconfig.h"
#include "dcmtk/dcmjpeg/djencext.h"
#include "dcmtk/dcmjpeg/djcparam.h"
#include "dcmtk/dcmjpeg/djrploss.h"
#include "dcmtk/dcmjpeg/djeijg8.h"
#include "dcmtk/dcmjpeg/djeijg12.h"

DJEncoderExtended::DJEncoderExtended()
: DJCodecEncoder()
{
}

DJEncoderExtended::~DJEncoderExtended()
{
}

E_TransferSyntax DJEncoderExtended::supportedTransferSyntax() const
{
  return EXS_JPEGProcess2_4;
}

OFBool DJEncoderExtended::isLosslessProcess() const
{
  return OFFalse;
}

void DJEncoderExtended::createDerivationDescription(
  const DcmRepresentationParameter * toRepParam,
  const DJCodecParameter * /* cp */ ,
  Uint8 bitsPerSample,
  double ratio,
  OFString& derivationDescription) const
{
  DJ_RPLossy defaultRP;
  const DJ_RPLossy *rp = toRepParam ? (const DJ_RPLossy *)toRepParam : &defaultRP ;
  char buf[64];

  derivationDescription =  "Lossy compression with JPEG extended sequential ";
  if (bitsPerSample > 8) derivationDescription += "12 bit"; else derivationDescription += "8 bit";
  derivationDescription += ", IJG quality factor ";
  sprintf(buf, "%u", rp->getQuality());
  derivationDescription += buf;
  derivationDescription += ", compression ratio ";
  appendCompressionRatio(derivationDescription, ratio);
}

DJEncoder *DJEncoderExtended::createEncoderInstance(
    const DcmRepresentationParameter * toRepParam,
    const DJCodecParameter *cp,
    Uint8 bitsPerSample) const
{
  DJ_RPLossy defaultRP;
  const DJ_RPLossy *rp = toRepParam ? (const DJ_RPLossy *)toRepParam : &defaultRP ;
  DJEncoder *result = NULL;
  if (bitsPerSample > 8)
    result = new DJCompressIJG12Bit(*cp, EJM_sequential, rp->getQuality());
  else
    result = new DJCompressIJG8Bit(*cp, EJM_sequential, rp->getQuality());
  return result;
}


text:
namespace mad {
namespace cppin {
namespace details {

template<typename Collection>
class in_impl_base {
public:
	typedef typename iterator_type<Collection>::type iterator;

	in_impl_base(const iterator& position, const iterator& end)
		: position_(position), end_(end) {}

	explicit operator bool() const {
		return end_ != position_;
	}

	iterator position() const {
		return position_;
	}
private:
	iterator position_;
	iterator end_;
};

template<typename Element, typename Collection, bool HasMemberFind = false>
class in_impl : public in_impl_base<Collection> {
public:
	in_impl(const Element& e, const Collection& c)
		: in_impl_base<Collection>(std::find(std::begin(c), std::end(c), e), std::end(c)) {}
};

template<typename Element, typename Collection>
class in_impl<Element, Collection, true> : public in_impl_base<Collection> {
public:
	in_impl(const Element& e, const Collection& c)
		: in_impl_base<Collection>(c.find(e), std::end(c)) {}
};

template<typename Element, typename Collection> using in_implementation = in_impl<Element, Collection, has_member_find<Collection>::value>;

template<typename Element>
class in_left {
public:
	in_left(const Element& e) : e_(e) {}

	template<typename Collection> friend
		in_implementation<Element, Collection> operator<<(const in_left<Element>& l,
											const Collection& c) {
		return in_implementation<Element, Collection>(l.e_, c);
	}
private:
	const Element& e_;
};

template<typename Collection>
class in_right {
public:
	in_right(const Collection& c) : c_(c) {}

	template<typename Element> friend
		in_implementation<Element, Collection> operator<<(const Element& e,
											const in_right<Collection>& r) {
		return in_implementation<Element, Collection>(e, r.c_);
	}
private:
	const Collection& c_;
};

}	// details

struct in_type {};

template<typename Element>
details::in_left<Element> operator<<(const Element& e, const in_type) {
	return details::in_left<Element>(e);
}

template<typename Collection>
details::in_right<Collection> operator<<(const in_type, const Collection& c) {
	return details::in_right<Collection>(c);
}

namespace {
const in_type in_;
}	// anon namespace
}	// cppin
}	// mad

#ifndef NO_IN_MACRO
#define in << ::mad::cppin::in_ <<
#endif


text:
set -e

# If this is a pull request then we won't have access to secure variables and can't do integration tests with SauceLabs.
#   In this case just do normal local tests
if [ $TRAVIS_PULL_REQUEST != "false" ]
then
  # Run default task
  gulp
else
  echo "travis_fold:start:Tests"
  gulp test
  echo "travis_fold:end:Tests"

  # Send coverage data to coveralls.io
  if [ $TRAVIS_BRANCH == "master" ]
  then
    cat ./.tmp/coverage/lcov.info | ./node_modules/coveralls/bin/coveralls.js || true
  fi

  gulp
fi


text:
<?php declare(strict_types=1);

namespace Monolog\Handler;

use Monolog\Test\TestCase;
use Monolog\Level;

class WhatFailureGroupHandlerTest extends TestCase
{
    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::__construct
     */
    public function testConstructorOnlyTakesHandler()
    {
        $this->expectException(\InvalidArgumentException::class);

        new WhatFailureGroupHandler([new TestHandler(), "foo"]);
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::__construct
     * @covers Monolog\Handler\WhatFailureGroupHandler::handle
     */
    public function testHandle()
    {
        $testHandlers = [new TestHandler(), new TestHandler()];
        $handler = new WhatFailureGroupHandler($testHandlers);
        $handler->handle($this->getRecord(Level::Debug));
        $handler->handle($this->getRecord(Level::Info));
        foreach ($testHandlers as $test) {
            $this->assertTrue($test->hasDebugRecords());
            $this->assertTrue($test->hasInfoRecords());
            $this->assertCount(2, $test->getRecords());
        }
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::handleBatch
     */
    public function testHandleBatch()
    {
        $testHandlers = [new TestHandler(), new TestHandler()];
        $handler = new WhatFailureGroupHandler($testHandlers);
        $handler->handleBatch([$this->getRecord(Level::Debug), $this->getRecord(Level::Info)]);
        foreach ($testHandlers as $test) {
            $this->assertTrue($test->hasDebugRecords());
            $this->assertTrue($test->hasInfoRecords());
            $this->assertCount(2, $test->getRecords());
        }
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::isHandling
     */
    public function testIsHandling()
    {
        $testHandlers = [new TestHandler(Level::Error), new TestHandler(Level::Warning)];
        $handler = new WhatFailureGroupHandler($testHandlers);
        $this->assertTrue($handler->isHandling($this->getRecord(Level::Error)));
        $this->assertTrue($handler->isHandling($this->getRecord(Level::Warning)));
        $this->assertFalse($handler->isHandling($this->getRecord(Level::Debug)));
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::handle
     */
    public function testHandleUsesProcessors()
    {
        $test = new TestHandler();
        $handler = new WhatFailureGroupHandler([$test]);
        $handler->pushProcessor(function ($record) {
            $record->extra['foo'] = true;

            return $record;
        });
        $handler->handle($this->getRecord(Level::Warning));
        $this->assertTrue($test->hasWarningRecords());
        $records = $test->getRecords();
        $this->assertTrue($records[0]['extra']['foo']);
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::handleBatch
     */
    public function testHandleBatchUsesProcessors()
    {
        $testHandlers = [new TestHandler(), new TestHandler()];
        $handler = new WhatFailureGroupHandler($testHandlers);
        $handler->pushProcessor(function ($record) {
            $record->extra['foo'] = true;

            return $record;
        });
        $handler->pushProcessor(function ($record) {
            $record->extra['foo2'] = true;

            return $record;
        });
        $handler->handleBatch([$this->getRecord(Level::Debug), $this->getRecord(Level::Info)]);
        foreach ($testHandlers as $test) {
            $this->assertTrue($test->hasDebugRecords());
            $this->assertTrue($test->hasInfoRecords());
            $this->assertCount(2, $test->getRecords());
            $records = $test->getRecords();
            $this->assertTrue($records[0]['extra']['foo']);
            $this->assertTrue($records[1]['extra']['foo']);
            $this->assertTrue($records[0]['extra']['foo2']);
            $this->assertTrue($records[1]['extra']['foo2']);
        }
    }

    /**
     * @covers Monolog\Handler\WhatFailureGroupHandler::handle
     */
    public function testHandleException()
    {
        $test = new TestHandler();
        $exception = new ExceptionTestHandler();
        $handler = new WhatFailureGroupHandler([$exception, $test, $exception]);
        $handler->pushProcessor(function ($record) {
            $record->extra['foo'] = true;

            return $record;
        });
        $handler->handle($this->getRecord(Level::Warning));
        $this->assertTrue($test->hasWarningRecords());
        $records = $test->getRecords();
        $this->assertTrue($records[0]['extra']['foo']);
    }
}


text:
package org.search.nibrs.stagingdata.model;

import javax.persistence.Cacheable;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

import org.apache.commons.lang3.builder.ToStringBuilder;

@Entity
@Cacheable
public class TypeOfVictimType {
	@Id
	@GeneratedValue(strategy = GenerationType.IDENTITY)
	private Integer typeOfVictimTypeId;

	private String stateCode;
	private String stateDescription;
	private String nibrsCode;
	private String nibrsDescription;

	public TypeOfVictimType() {
		super();
	}

	public TypeOfVictimType(Integer typeOfVictimTypeId) {
		super();
		this.typeOfVictimTypeId = typeOfVictimTypeId;
	}

	public TypeOfVictimType(Integer typeOfVictimTypeId, String stateCode, String stateDescription, String nibrsCode,
			String nibrsDescription) {
		super();
		this.typeOfVictimTypeId = typeOfVictimTypeId;
		this.stateCode = stateCode;
		this.stateDescription = stateDescription;
		this.nibrsCode = nibrsCode;
		this.nibrsDescription = nibrsDescription;
	}

	public String toString(){
		return ToStringBuilder.reflectionToString(this);
	}

	public Integer getTypeOfVictimTypeId() {
		return typeOfVictimTypeId;
	}

	public void setTypeOfVictimTypeId(Integer typeOfVictimTypeId) {
		this.typeOfVictimTypeId = typeOfVictimTypeId;
	}

	public String getStateCode() {
		return stateCode;
	}

	public void setStateCode(String stateCode) {
		this.stateCode = stateCode;
	}

	public String getStateDescription() {
		return stateDescription;
	}

	public void setStateDescription(String stateDescription) {
		this.stateDescription = stateDescription;
	}

	public String getNibrsCode() {
		return nibrsCode;
	}

	public void setNibrsCode(String nibrsCode) {
		this.nibrsCode = nibrsCode;
	}

	public String getNibrsDescription() {
		return nibrsDescription;
	}

	public void setNibrsDescription(String nibrsDescription) {
		this.nibrsDescription = nibrsDescription;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((nibrsCode == null) ? 0 : nibrsCode.hashCode());
		result = prime * result + ((nibrsDescription == null) ? 0 : nibrsDescription.hashCode());
		result = prime * result + ((stateCode == null) ? 0 : stateCode.hashCode());
		result = prime * result + ((stateDescription == null) ? 0 : stateDescription.hashCode());
		result = prime * result + ((typeOfVictimTypeId == null) ? 0 : typeOfVictimTypeId.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		TypeOfVictimType other = (TypeOfVictimType) obj;
		if (nibrsCode == null) {
			if (other.nibrsCode != null)
				return false;
		} else if (!nibrsCode.equals(other.nibrsCode))
			return false;
		if (nibrsDescription == null) {
			if (other.nibrsDescription != null)
				return false;
		} else if (!nibrsDescription.equals(other.nibrsDescription))
			return false;
		if (stateCode == null) {
			if (other.stateCode != null)
				return false;
		} else if (!stateCode.equals(other.stateCode))
			return false;
		if (stateDescription == null) {
			if (other.stateDescription != null)
				return false;
		} else if (!stateDescription.equals(other.stateDescription))
			return false;
		if (typeOfVictimTypeId == null) {
			if (other.typeOfVictimTypeId != null)
				return false;
		} else if (!typeOfVictimTypeId.equals(other.typeOfVictimTypeId))
			return false;
		return true;
	}

}


text:
package org.fossasia.pslab.communication.peripherals;

import android.util.Log;

import org.fossasia.pslab.communication.SensorList;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;

/**
 * Created by viveksb007 on 28/3/17.
 */

public class RadioLink {

    private static final String TAG = "RadioLink";
    private int ADC_COMMANDS = 1;
    private int READ_ADC = 0 << 4;

    private int I2C_COMMANDS = 2;
    private int I2C_TRANSACTION = 0 << 4;
    private int I2C_WRITE = 1 << 4;
    private int I2C_SCAN = 2 << 4;
    private int PULL_SCL_LOW = 3 << 4;
    private int I2C_CONFIG = 4 << 4;
    private int I2C_READ = 5 << 4;

    private int NRF_COMMANDS = 3;
    private int NRF_READ_REGISTER = 0 << 4;
    private int NRF_WRITE_REGISTER = 1 << 4;

    private int MISC_COMMANDS = 4;
    private int WS2812B_CMD = 0 << 4;

    private NRF24L01 nrf24L01;
    private int ADDRESS = 0x010101;
    private int timeout = 200;

    public RadioLink(NRF24L01 nrf24L01, int address) {
        this.nrf24L01 = nrf24L01;
        if (address != -1) ADDRESS = address;

    }

    public void selectMe() {
        if (this.nrf24L01.CURRENT_ADDRESS != this.ADDRESS) {
            this.nrf24L01.selectAddress(this.ADDRESS);
        }
    }

    public ArrayList<Character> writeI2C(int I2CAddress, int regAddress, int[] data) throws IOException {
        selectMe();
        int[] newData = new int[3 + data.length];
        newData[0] = I2C_COMMANDS | I2C_WRITE;
        newData[1] = I2CAddress;
        newData[2] = regAddress;
        System.arraycopy(data, 0, newData, 3, data.length);
        return this.nrf24L01.transaction(newData, 0, timeout);
    }

    public ArrayList<Character> readI2C(int I2CAddress, int regAddress, int numBytes) throws IOException {
        selectMe();
        return this.nrf24L01.transaction(new int[]{I2C_COMMANDS | I2C_TRANSACTION, I2CAddress, regAddress, numBytes}, 0, timeout);
    }

    public ArrayList<Character> writeBulk(int I2CAddress, int[] data) throws IOException {
        selectMe();
        int[] newData = new int[2 + data.length];
        newData[0] = I2C_COMMANDS | I2C_WRITE;
        newData[1] = I2CAddress;
        System.arraycopy(data, 0, newData, 2, data.length);
        return this.nrf24L01.transaction(newData, 0, timeout);
    }

    public ArrayList<Character> readBulk(int I2CAddress, int regAddress, int numBytes) throws IOException {
        selectMe();
        return this.nrf24L01.transactionWithRetries(new int[]{I2C_COMMANDS | I2C_TRANSACTION, I2CAddress, regAddress, numBytes}, -1);
    }

    public ArrayList<Character> simpleRead(int I2CAddress, int numBytes) throws IOException {
        selectMe();
        return this.nrf24L01.transactionWithRetries(new int[]{I2C_COMMANDS | I2C_READ, I2CAddress, numBytes}, -1);
    }

    public ArrayList<Character> readADC(int channel) throws IOException {
        selectMe();
        return this.nrf24L01.transaction(new int[]{ADC_COMMANDS | READ_ADC, channel}, 0, timeout);
    }

    public ArrayList<Integer> i2CScan() throws IOException {
        selectMe();
        Log.v(TAG, "Scanning addresses 0-127...");
        ArrayList<Integer> addresses = new ArrayList<>();
        ArrayList<Character> temp = this.nrf24L01.transaction(new int[]{I2C_COMMANDS | I2C_SCAN | 0x80}, 0, 500);
        if (temp == null) return addresses;
        int sum = 0;
        for (int i = 0; i < temp.size(); i++) {
            sum += (int) temp.get(i);
        }
        if (sum == 0) return addresses;
        Log.v(TAG, "Address \t Possible Devices");
        SensorList sensorList = new SensorList();
        for (int i = 0; i < 16; i++) {
            if ((temp.get(i) ^ 255) != 0) {
                for (int j = 0; j < 8; j++) {
                    if ((temp.get(i) & (0x80 >> j)) == 0) {
                        int address = 8 * i + j;
                        addresses.add(address);
                        Log.v(TAG, Integer.toHexString(address) + "\t" + Arrays.toString(sensorList.sensorList.get(address)));
                    }
                }
            }
        }
        return addresses;
    }

    public ArrayList<Integer> decodeI2CList(int[] data) {
        int sum = 0;
        ArrayList<Integer> addressList = new ArrayList<>();
        for (int _data : data) {
            sum += _data;
        }
        if (sum == 0) return addressList;
        for (int i = 0; i < data.length; i++) {
            if ((data[i] ^ 255) != 0) {
                for (int j = 0; j < 8; j++) {
                    if ((data[i] & (0x80 >> j)) == 0) {
                        addressList.add(8 * i + j);
                    }
                }
            }
        }
        return addressList;
    }

    public ArrayList<Integer> pullSCLLow(int t_ms) throws IOException {
        selectMe();
        ArrayList<Character> data;
        data = this.nrf24L01.transaction(new int[]{I2C_COMMANDS | PULL_SCL_LOW, t_ms}, 0, timeout);
        if (data != null) {
            int[] tempData = new int[data.size()];
            for (int i = 0; i < data.size(); i++) {
                tempData[i] = data.get(i);
            }
            return decodeI2CList(tempData);
        } else {
            return new ArrayList<>();
        }
    }

    public ArrayList<Character> configI2C(int frequency) throws IOException {
        selectMe();
        int brgVal = (int) 32e6 / frequency / 4 - 1;
        return this.nrf24L01.transaction(new int[]{I2C_COMMANDS | I2C_CONFIG, brgVal}, 0, timeout);
    }

    public ArrayList<Character> writeRegister(int register, int value) throws IOException {
        selectMe();
        return this.nrf24L01.transaction(new int[]{NRF_COMMANDS | NRF_WRITE_REGISTER, register, value}, 0, timeout);
    }

    public int readRegister(int register) throws IOException {
        selectMe();
        ArrayList<Character> data = this.nrf24L01.transaction(new int[]{NRF_COMMANDS | NRF_READ_REGISTER, register}, 0, timeout);
        if (data != null) {
            return (int) data.get(0);
        } else
            return -1;
    }

    public ArrayList<Character> WS2812B(int[][] cols) throws IOException {
        selectMe();
        int[] colorArray = new int[1 + cols.length * 3];
        colorArray[0] = MISC_COMMANDS | WS2812B_CMD;
        for (int i = 1, j = 1; i < cols.length; i++, j += 3) {
            colorArray[j] = cols[i][1];
            colorArray[j + 1] = cols[i][0];
            colorArray[j + 2] = cols[i][2];
        }
        return this.nrf24L01.transaction(colorArray, 0, timeout);
    }

}


text:
using System;
using System.Linq;
using BenchmarkDotNet.Characteristics;
using BenchmarkDotNet.Environments;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Reports;
using BenchmarkDotNet.Running;

namespace BenchmarkDotNet.Columns
{
    public class JobCharacteristicColumn : IColumn
    {
        private static readonly CharacteristicPresenter Presenter = CharacteristicPresenter.SummaryPresenter;

        private readonly Characteristic characteristic;

        public JobCharacteristicColumn(Characteristic characteristic)
        {
            this.characteristic = characteristic;
            Id = "Job." + characteristic.Id;
            ColumnName = characteristic.Id;

            // The 'Id' characteristic is a special case:
            // here we just print 'Job'
            if (characteristic.Id == "Id")
                ColumnName = "Job";
        }

        public string Id { get; }
        public string ColumnName { get; }
        public bool IsAvailable(Summary summary) => true;
        public bool AlwaysShow => false;
        public ColumnCategory Category => ColumnCategory.Job;
        public int PriorityInCategory => 0;
        public bool IsNumeric => false;
        public UnitType UnitType => UnitType.Dimensionless;
        public string GetValue(Summary summary, Benchmark benchmark, ISummaryStyle style) => GetValue(summary, benchmark);

        public bool IsDefault(Summary summary, Benchmark benchmark) => !benchmark.Job.HasValue(characteristic);

        public string GetValue(Summary summary, Benchmark benchmark)
        {
            if (!benchmark.Job.HasValue(characteristic) && EnvResolver.Instance.CanResolve(characteristic))
                return benchmark.Job.ResolveValue(characteristic, EnvResolver.Instance).ToString();
            return Presenter.ToPresentation(benchmark.Job, characteristic);
        }

        private static readonly Lazy<IColumn[]> LazyAllColumns =
            new Lazy<IColumn[]>(() =>
                CharacteristicHelper.GetAllPresentableCharacteristics(typeof(Job), true)
                .Select(c => (IColumn)new JobCharacteristicColumn(c)).ToArray());

        public static IColumn[] AllColumns => LazyAllColumns.Value;

        public string Legend => "";
    }
}


text:
Working with sabre rest api trough laravel.

Laravel package for https://github.com/SabreDevStudio/SabrePhpBridge.


text:
package li.strolch.soql.core.expresssion;

import java.util.Map;

/**
 * @author msmock
 */
public interface IBooleanExpression extends IExpression {

	boolean evaluate(Map<String, Object> inputObjects, Map<String, Object> queryParameter);

}


text:
WPA_DEFINES = -DCONFIG_CTRL_IFACE -DCONFIG_CTRL_IFACE_UNIX

LDFLAGS += -lrt
CFLAGS ?= -O2 -Wall -Wextra -Wno-unused-parameter
CC ?= $(CROSSCOMPILE)gcc
MIX ?= mix

# If not cross-compiling, then run sudo by default
ifeq ($(origin CROSSCOMPILE), undefined)
SUDO_ASKPASS ?= /usr/bin/ssh-askpass
SUDO ?= sudo
else
# If cross-compiling, then permissions need to be set some build system-dependent way
SUDO ?= true
endif

all: compile

compile:
	$(MIX) compile

test:
	$(MIX) test

%.o: %.c
	$(CC) -c $(WPA_DEFINES) $(CFLAGS) -o $@ $<

priv/wpa_ex: src/wpa_ex.o src/wpa_ctrl/os_unix.o src/wpa_ctrl/wpa_ctrl.o
	@mkdir -p priv
	$(CC) $^ $(LDFLAGS) -o $@
	# setuid root wpa_ex so that it can interact with the wpa_supplicant
	SUDO_ASKPASS=$(SUDO_ASKPASS) $(SUDO) -- sh -c 'chown root:root $@; chmod +s $@'

clean:
	$(MIX) clean
	rm -f priv/wpa_ex src/*.o src/wpa_ctrl/*.o

.PHONY: all compile test clean


text:
<!DOCTYPE html>

<title>개인설정 - 비밀번호</title>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />

<base href="/" />

	<md-content class="content-container content" ng-controller="settings.passwordController">

			개인설정
			비밀번호

		<ui-view />
	</md-content>


text:
from torchvision.datasets import CIFAR100
import numpy as np

class CIFAR100N(CIFAR100):
    """
    Extends CIFAR100 dataset to yield index of element in addition to image and target label.
    """

    def __init__(self,
                 root,
                 train=True,
                 transform=None,
                 target_transform=None,
                 download=False,
                 rand_fraction=0.0):
        super(CIFAR100N, self).__init__(root=root,
                                              train=train,
                                              transform=transform,
                                              target_transform=target_transform,
                                              download=download)

        assert (rand_fraction <= 1.0) and (rand_fraction >= 0.0)
        self.rand_fraction = rand_fraction

        if self.rand_fraction > 0.0:
            self.targets = self.corrupt_fraction_of_data()

    def corrupt_fraction_of_data(self):
        """Corrupts fraction of train data by permuting image-label pairs."""

        # Check if we are not corrupting test data
        assert self.train is True, 'We should not corrupt test data.'
        rearrange = []
        length = len(self.targets)//4
        start_next = 0
        new_labels = []
        for i in range(4):
            nr_corrupt_instances = start_next + int(np.floor(length * self.rand_fraction))
            print('Randomizing {} fraction of data == {} / {}'.format(self.rand_fraction,
                                                                      nr_corrupt_instances-start_next ,
                                                                      length))
            # We will corrupt the top fraction data points
            corrupt_label = self.targets[start_next:nr_corrupt_instances]
            clean_label = self.targets[nr_corrupt_instances:start_next + length]

            # Corrupting data
            np.random.seed(111)
            rand_idx = np.random.permutation(np.arange(start_next,nr_corrupt_instances))
            corrupt_label = np.array(corrupt_label)[rand_idx-start_next]
            # Adding corrupt and clean data back together
            new_labels.extend(corrupt_label)
            new_labels.extend(clean_label)
            start_next += length

        return np.array(new_labels)

    def __getitem__(self, index):
        """
        Args:
            index (int):  index of element to be fetched

        Returns:
            tuple: (sample, target, index) where index is the index of this sample in dataset.
        """
        img, target = super().__getitem__(index)
        #return img, target, index
        return img, target


text:
package org.dasein.cloud.aws.storage;

import org.dasein.cloud.CloudErrorType;
import org.dasein.cloud.CloudException;

public class S3Exception extends CloudException {
	private static final long serialVersionUID = -1187862739180492610L;

	private String requestId = null;

	public S3Exception(int status, String requestId, String code, String message) {
		super(CloudErrorType.GENERAL, status, code, message);
		this.requestId = requestId;
	}

	public String getCode() {
		return getProviderCode();
	}

	public String getRequestId() {
		return requestId;
	}

	public int getStatus() {
		return getHttpCode();
	}

	public String getSummary() {
		return (getStatus() + "/" + requestId + "/" + getCode() + ": " + getMessage());
	}
}


text:
<?php
/**
 * View class
 */

class View
{
	private static $smarty = false;
	private static $instance = False;

	private function __construct()
	{
		//self:$smarty->force_compile = true;
		//Smarty related obj and settings
		self::$smarty = new Smarty();

		self::$smarty->compile_check = true;
		self::$smarty->debugging = false;
		self::$smarty->caching = false;
		self::$smarty->cache_lifetime = 120;

		//Note: these dirs have to be writeable
		self::$smarty->setCacheDir(Config::DIR.'/libs/smarty/cache/');
		self::$smarty->setCompileDir(Config::DIR.'/libs/smarty/templates_c/');
	}

	/**
     * Get the singleton instance
     *
     * @return View
     *
    */
    public static function &get()
    {
    	if (!self::$instance)
    	{
    		self::$instance = new View();
    	}

    	return self::$instance;
    }

	public function assign( $name, $value )
	{
		self::$smarty->assign( $name, $value );
	}

	public function fetch( $tpl )
	{
		$res = self::$smarty->fetch( $tpl );
		return $res;
	}

	public function display()
	{
		//Assign ending microtime on pageload, NOTE: this is the last call of the framework, before page rendering
		$this->assign("end_time",microtime( true ));

		//Display the page
		self::$smarty->display(Config::DIR."/moduls/templates/".Config::$VARIABLES['TEMPLATE']."/_page.tpl");
	}

}