/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.ozone.container.ec.reconstruction;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.ContainerClientMetrics;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
import org.apache.hadoop.hdds.scm.StreamBufferArgs;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
import org.apache.hadoop.hdds.scm.storage.BufferPool;
import org.apache.hadoop.hdds.scm.storage.ECBlockOutputStream;
import org.apache.hadoop.hdds.security.SecurityConfig;
import org.apache.hadoop.hdds.security.symmetric.SecretKeySignerClient;
import org.apache.hadoop.hdds.security.token.ContainerTokenIdentifier;
import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
import org.apache.hadoop.hdds.utils.IOUtils;
import org.apache.hadoop.io.ByteBufferPool;
import org.apache.hadoop.io.ElasticByteBufferPool;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.client.io.BlockInputStreamFactory;
import org.apache.hadoop.ozone.client.io.BlockInputStreamFactoryImpl;
import org.apache.hadoop.ozone.client.io.ECBlockInputStreamProxy;
import org.apache.hadoop.ozone.client.io.ECBlockReconstructedStripeInputStream;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
import org.apache.hadoop.security.token.Token;
import org.apache.ratis.util.MemoizedSupplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.apache.hadoop.ozone.container.ec.reconstruction.TokenHelper.encode;

/**
 * The Coordinator implements the main flow of reconstructing
 * missing container replicas.
 * <p>
 * For a container reconstruction task, the main flow is:
 * - ListBlock from all healthy replicas
 * - calculate effective block group len for all blocks
 * - create RECOVERING containers in TargetDNs
 * -  for each block
 * -    build a ECReconstructedStripedInputStream to read healthy chunks
 * -    build a ECBlockOutputStream to write out decoded chunks
 * -      for each stripe
 * -        use ECReconstructedStripedInputStream.recoverChunks to decode chunks
 * -        use ECBlockOutputStream.write to write decoded chunks to TargetDNs
 * -    PutBlock
 * - Close RECOVERING containers in TargetDNs
 */
public class ECReconstructionCoordinator implements Closeable {

  static final Logger LOG =
      LoggerFactory.getLogger(ECReconstructionCoordinator.class);

  private static final int EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE = 3;

  private static final int EC_RECONSTRUCT_STRIPE_WRITE_POOL_MIN_SIZE = 5;

  private final ECContainerOperationClient containerOperationClient;

  private final ByteBufferPool byteBufferPool;

  private final ExecutorService ecReconstructReadExecutor;
  private final MemoizedSupplier<ExecutorService> ecReconstructWriteExecutor;
  private final BlockInputStreamFactory blockInputStreamFactory;
  private final TokenHelper tokenHelper;
  private final ContainerClientMetrics clientMetrics;
  private final ECReconstructionMetrics metrics;
  private final StateContext context;
  private final OzoneClientConfig ozoneClientConfig;

  public ECReconstructionCoordinator(
      ConfigurationSource conf, CertificateClient certificateClient,
      SecretKeySignerClient secretKeyClient, StateContext context,
      ECReconstructionMetrics metrics,
      String threadNamePrefix) throws IOException {
    this.context = context;
    this.containerOperationClient = new ECContainerOperationClient(conf,
        certificateClient);
    this.byteBufferPool = new ElasticByteBufferPool();
    ozoneClientConfig = conf.getObject(OzoneClientConfig.class);
    this.ecReconstructReadExecutor = createThreadPoolExecutor(
        EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE,
        ozoneClientConfig.getEcReconstructStripeReadPoolLimit(),
        threadNamePrefix + "ec-reconstruct-reader-TID-%d");
    this.ecReconstructWriteExecutor = MemoizedSupplier.valueOf(
        () -> createThreadPoolExecutor(
            EC_RECONSTRUCT_STRIPE_WRITE_POOL_MIN_SIZE,
            ozoneClientConfig.getEcReconstructStripeWritePoolLimit(),
            threadNamePrefix + "ec-reconstruct-writer-TID-%d"));
    this.blockInputStreamFactory = BlockInputStreamFactoryImpl
        .getInstance(byteBufferPool, () -> ecReconstructReadExecutor);
    tokenHelper = new TokenHelper(new SecurityConfig(conf), secretKeyClient);
    this.clientMetrics = ContainerClientMetrics.acquire();
    this.metrics = metrics;
  }

  public void reconstructECContainerGroup(long containerID,
      ECReplicationConfig repConfig,
      SortedMap<Integer, DatanodeDetails> sourceNodeMap,
      SortedMap<Integer, DatanodeDetails> targetNodeMap) throws IOException {

    Pipeline pipeline = rebuildInputPipeline(repConfig, sourceNodeMap);

    SortedMap<Long, BlockData[]> blockDataMap =
        getBlockDataMap(containerID, repConfig, sourceNodeMap);

    SortedMap<Long, BlockLocationInfo> blockLocationInfoMap =
        calcBlockLocationInfoMap(containerID, blockDataMap, pipeline);
    ContainerID cid = ContainerID.valueOf(containerID);

    // 1. create target recovering containers.
    String containerToken = encode(tokenHelper.getContainerToken(cid));
    List<DatanodeDetails> recoveringContainersCreatedDNs = new ArrayList<>();
    try {
      for (Map.Entry<Integer, DatanodeDetails> indexDnPair : targetNodeMap
          .entrySet()) {
        DatanodeDetails dn = indexDnPair.getValue();
        int index = indexDnPair.getKey();
        LOG.debug("Creating container {} on datanode {} for index {}",
            containerID, dn, index);
        containerOperationClient
            .createRecoveringContainer(containerID, dn, repConfig,
                containerToken, index);
        recoveringContainersCreatedDNs.add(dn);
      }

      // 2. Reconstruct and transfer to targets
      for (Map.Entry<Long, BlockLocationInfo> blockLocationInfoEntry
          : blockLocationInfoMap.entrySet()) {
        Long key = blockLocationInfoEntry.getKey();
        BlockLocationInfo blockLocationInfo = blockLocationInfoEntry.getValue();
        reconstructECBlockGroup(blockLocationInfo, repConfig,
            targetNodeMap, blockDataMap.get(key));
      }

      // 3. Close containers
      for (DatanodeDetails dn: recoveringContainersCreatedDNs) {
        LOG.debug("Closing container {} on datanode {}", containerID, dn);
        containerOperationClient
            .closeContainer(containerID, dn, repConfig, containerToken);
      }
      metrics.incReconstructionTotal();
      metrics.incBlockGroupReconstructionTotal(blockLocationInfoMap.size());
    } catch (Exception e) {
      // Any exception let's delete the recovering containers.
      metrics.incReconstructionFailsTotal();
      metrics.incBlockGroupReconstructionFailsTotal(
          blockLocationInfoMap.size());
      LOG.warn(
          "Exception while reconstructing the container {}. Cleaning up"
              + " all the recovering containers in the reconstruction process.",
          containerID, e);
      // Delete only the current thread successfully created recovering
      // containers.
      for (DatanodeDetails dn : recoveringContainersCreatedDNs) {
        try {
          containerOperationClient
              .deleteContainerInState(containerID, dn, repConfig,
                  containerToken, ImmutableSet.of(
                          ContainerProtos.ContainerDataProto.State.UNHEALTHY,
                          ContainerProtos.ContainerDataProto.State.RECOVERING));
          if (LOG.isDebugEnabled()) {
            LOG.debug("Deleted the container {}, at the target: {}",
                containerID, dn);
          }
        } catch (IOException ioe) {
          LOG.error("Exception while deleting the container {} at target: {}",
              containerID, dn, ioe);
        }
      }
      throw e;
    }

  }

  private ECBlockOutputStream getECBlockOutputStream(
      BlockLocationInfo blockLocationInfo, DatanodeDetails datanodeDetails,
      ECReplicationConfig repConfig, int replicaIndex) throws IOException {
    StreamBufferArgs streamBufferArgs =
        StreamBufferArgs.getDefaultStreamBufferArgs(repConfig, ozoneClientConfig);
    return new ECBlockOutputStream(
        blockLocationInfo.getBlockID(),
        containerOperationClient.getXceiverClientManager(),
        containerOperationClient.singleNodePipeline(datanodeDetails,
            repConfig, replicaIndex),
        BufferPool.empty(), ozoneClientConfig,
        blockLocationInfo.getToken(), clientMetrics, streamBufferArgs, ecReconstructWriteExecutor);
  }

  @VisibleForTesting
  public void reconstructECBlockGroup(BlockLocationInfo blockLocationInfo,
      ECReplicationConfig repConfig,
      SortedMap<Integer, DatanodeDetails> targetMap, BlockData[] blockDataGroup)
      throws IOException {
    long safeBlockGroupLength = blockLocationInfo.getLength();
    List<Integer> missingContainerIndexes = new ArrayList<>(targetMap.keySet());

    // calculate the real missing block indexes
    int dataLocs = ECBlockInputStreamProxy
        .expectedDataLocations(repConfig, safeBlockGroupLength);
    List<Integer> toReconstructIndexes = new ArrayList<>();
    List<Integer> notReconstructIndexes = new ArrayList<>();
    for (Integer index : missingContainerIndexes) {
      if (index <= dataLocs || index > repConfig.getData()) {
        toReconstructIndexes.add(index);
      } else {
        // Don't need to be reconstructed, but we do need a stream to write
        // the block data to.
        notReconstructIndexes.add(index);
      }
    }

    OzoneClientConfig clientConfig = this.ozoneClientConfig;
    clientConfig.setChecksumVerify(true);
    try (ECBlockReconstructedStripeInputStream sis
        = new ECBlockReconstructedStripeInputStream(
        repConfig, blockLocationInfo,
        this.containerOperationClient.getXceiverClientManager(), null,
        this.blockInputStreamFactory, byteBufferPool,
        this.ecReconstructReadExecutor,
        clientConfig)) {

      ECBlockOutputStream[] targetBlockStreams =
          new ECBlockOutputStream[toReconstructIndexes.size()];
      ECBlockOutputStream[] emptyBlockStreams =
          new ECBlockOutputStream[notReconstructIndexes.size()];
      ByteBuffer[] bufs = new ByteBuffer[toReconstructIndexes.size()];
      try {
        // Create streams and buffers for all indexes that need reconstructed
        for (int i = 0; i < toReconstructIndexes.size(); i++) {
          int replicaIndex = toReconstructIndexes.get(i);
          DatanodeDetails datanodeDetails = targetMap.get(replicaIndex);
          targetBlockStreams[i] = getECBlockOutputStream(blockLocationInfo, datanodeDetails, repConfig, replicaIndex);
          bufs[i] = byteBufferPool.getBuffer(false, repConfig.getEcChunkSize());
          bufs[i].clear();
        }
        // Then create a stream for all indexes that don't need reconstructed, but still need a stream to
        // write the empty block data to.
        for (int i = 0; i < notReconstructIndexes.size(); i++) {
          int replicaIndex = notReconstructIndexes.get(i);
          DatanodeDetails datanodeDetails = targetMap.get(replicaIndex);
          emptyBlockStreams[i] = getECBlockOutputStream(blockLocationInfo, datanodeDetails, repConfig, replicaIndex);
        }

        if (toReconstructIndexes.size() > 0) {
          sis.setRecoveryIndexes(toReconstructIndexes.stream().map(i -> (i - 1))
              .collect(Collectors.toSet()));
          long length = safeBlockGroupLength;
          while (length > 0) {
            int readLen;
            try {
              readLen = sis.recoverChunks(bufs);
              Set<Integer> failedIndexes = sis.getFailedIndexes();
              if (!failedIndexes.isEmpty()) {
                // There was a problem reading some of the block indexes, but we
                // did not get an exception as there must have been spare indexes
                // to try and recover from. Therefore we should log out the block
                // group details in the same way as for the exception case below.
                logBlockGroupDetails(blockLocationInfo, repConfig,
                    blockDataGroup);
              }
            } catch (IOException e) {
              // When we see exceptions here, it could be due to some transient
              // issue that causes the block read to fail when reconstructing it,
              // but we have seen issues where the containers don't have the
              // blocks they appear they should have, or the block chunks are the
              // wrong length etc. In order to debug these sort of cases, if we
              // get an error, we will log out the details about the block group
              // length on each source, along with their chunk list and chunk
              // lengths etc.
              logBlockGroupDetails(blockLocationInfo, repConfig,
                  blockDataGroup);
              throw e;
            }
            // TODO: can be submitted in parallel
            for (int i = 0; i < bufs.length; i++) {
              CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
                  future = targetBlockStreams[i].write(bufs[i]);
              checkFailures(targetBlockStreams[i], future);
              bufs[i].clear();
            }
            length -= readLen;
          }
        }
        List<ECBlockOutputStream> allStreams = new ArrayList<>(Arrays.asList(targetBlockStreams));
        allStreams.addAll(Arrays.asList(emptyBlockStreams));
        for (ECBlockOutputStream targetStream : allStreams) {
          targetStream.executePutBlock(true, true, blockLocationInfo.getLength(), blockDataGroup);
          checkFailures(targetStream, targetStream.getCurrentPutBlkResponseFuture());
        }
      } finally {
        for (ByteBuffer buf : bufs) {
          byteBufferPool.putBuffer(buf);
        }
        IOUtils.cleanupWithLogger(LOG, targetBlockStreams);
        IOUtils.cleanupWithLogger(LOG, emptyBlockStreams);
      }
    }
  }

  private void logBlockGroupDetails(BlockLocationInfo blockLocationInfo,
      ECReplicationConfig repConfig, BlockData[] blockDataGroup) {
    LOG.info("Block group details for {}. " +
        "Replication Config {}. Calculated safe length: {}. ",
        blockLocationInfo.getBlockID(), repConfig,
        blockLocationInfo.getLength());
    for (int i = 0; i < blockDataGroup.length; i++) {
      BlockData data = blockDataGroup[i];
      if (data == null) {
        continue;
      }
      StringBuilder sb = new StringBuilder();
      sb.append("Block Data for: ")
          .append(data.getBlockID())
          .append(" replica Index: ")
          .append(i + 1)
          .append(" block length: ")
          .append(data.getSize())
          .append(" block group length: ")
          .append(getBlockDataLength(data))
          .append(" chunk list: \n");
      int cnt = 0;
      for (ContainerProtos.ChunkInfo chunkInfo : data.getChunks()) {
        if (cnt > 0) {
          sb.append("\n");
        }
        sb.append("  chunkNum: ")
            .append(++cnt)
            .append(" length: ")
            .append(chunkInfo.getLen())
            .append(" offset: ")
            .append(chunkInfo.getOffset());
      }
      LOG.info(sb.toString());
    }
  }

  private void checkFailures(ECBlockOutputStream targetBlockStream,
      CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
          currentPutBlkResponseFuture)
      throws IOException {
    if (isFailed(targetBlockStream, currentPutBlkResponseFuture)) {
      // If one chunk response failed, we should retry.
      // Even after retries if it failed, we should declare the
      // reconstruction as failed.
      // For now, let's throw the exception.
      throw new IOException("Chunk write failed at the new target node: " +
          targetBlockStream.getDatanodeDetails() +
          ". Aborting the reconstruction process.",
          targetBlockStream.getIoException());
    }
  }

  private boolean isFailed(ECBlockOutputStream outputStream,
      CompletableFuture<ContainerProtos.
          ContainerCommandResponseProto> chunkWriteResponseFuture) {
    if (chunkWriteResponseFuture == null) {
      return true;
    }

    ContainerProtos.ContainerCommandResponseProto
        containerCommandResponseProto = null;
    try {
      containerCommandResponseProto = chunkWriteResponseFuture.get();
    } catch (InterruptedException e) {
      outputStream.setIoException(e);
      Thread.currentThread().interrupt();
    } catch (ExecutionException e) {
      outputStream.setIoException(e);
    }

    if (outputStream.getIoException() != null) {
      return true;
    }

    if (containerCommandResponseProto == null) {
      return true;
    }

    return false;
  }

  SortedMap<Long, BlockLocationInfo> calcBlockLocationInfoMap(long containerID,
      SortedMap<Long, BlockData[]> blockDataMap, Pipeline pipeline) {

    SortedMap<Long, BlockLocationInfo> blockInfoMap = new TreeMap<>();

    for (Map.Entry<Long, BlockData[]> entry : blockDataMap.entrySet()) {
      Long localID = entry.getKey();
      BlockData[] blockGroup = entry.getValue();

      long blockGroupLen = calcEffectiveBlockGroupLen(blockGroup,
          pipeline.getReplicationConfig().getRequiredNodes());
      if (blockGroupLen > 0) {
        BlockID blockID = new BlockID(containerID, localID);
        BlockLocationInfo blockLocationInfo = new BlockLocationInfo.Builder()
            .setBlockID(blockID)
            .setLength(blockGroupLen)
            .setPipeline(pipeline)
            .setToken(tokenHelper.getBlockToken(blockID, blockGroupLen))
            .build();
        blockInfoMap.put(localID, blockLocationInfo);
      }
    }
    return blockInfoMap;
  }

  @Override
  public void close() throws IOException {
    if (containerOperationClient != null) {
      containerOperationClient.close();
    }
    if (ecReconstructWriteExecutor.isInitialized()) {
      ecReconstructWriteExecutor.get().shutdownNow();
    }
  }

  private Pipeline rebuildInputPipeline(ECReplicationConfig repConfig,
      SortedMap<Integer, DatanodeDetails> sourceNodeMap) {

    List<DatanodeDetails> nodes = new ArrayList<>(sourceNodeMap.values());
    Map<DatanodeDetails, Integer> dnVsIndex = new HashMap<>();

    Iterator<Map.Entry<Integer, DatanodeDetails>> iterator =
        sourceNodeMap.entrySet().iterator();
    while (iterator.hasNext()) {
      Map.Entry<Integer, DatanodeDetails> next = iterator.next();
      Integer key = next.getKey();
      DatanodeDetails value = next.getValue();
      dnVsIndex.put(value, key);
    }

    return Pipeline.newBuilder().setId(PipelineID.randomId())
        .setReplicationConfig(repConfig).setNodes(nodes)
        .setReplicaIndexes(dnVsIndex).setState(Pipeline.PipelineState.CLOSED)
        .build();
  }

  private SortedMap<Long, BlockData[]> getBlockDataMap(long containerID,
      ECReplicationConfig repConfig,
      Map<Integer, DatanodeDetails> sourceNodeMap) throws IOException {

    SortedMap<Long, BlockData[]> resultMap = new TreeMap<>();
    Token<ContainerTokenIdentifier> containerToken =
        tokenHelper.getContainerToken(new ContainerID(containerID));

    Iterator<Map.Entry<Integer, DatanodeDetails>> iterator =
        sourceNodeMap.entrySet().iterator();

    while (iterator.hasNext()) {
      Map.Entry<Integer, DatanodeDetails> next = iterator.next();
      Integer index = next.getKey();
      DatanodeDetails dn = next.getValue();

      BlockData[] blockDataArr = containerOperationClient.listBlock(
          containerID, dn, repConfig, containerToken);

      for (BlockData blockData : blockDataArr) {
        BlockID blockID = blockData.getBlockID();
        BlockData[] blkDataArr = resultMap.getOrDefault(blockData.getLocalID(),
            new BlockData[repConfig.getRequiredNodes()]);
        blkDataArr[index - 1] = blockData;
        resultMap.put(blockID.getLocalID(), blkDataArr);
      }
    }
    // When a stripe is written, the put block is sent to all nodes even if
    // that nodes has zero bytes written to it. If the
    // client does not get an ACK from all nodes, it will abandon the stripe,
    // which can leave incomplete stripes on the DNs. Therefore, we should check
    // that all blocks in the result map have an entry for all nodes. If they
    // do not, it means this is an abandoned stripe and we should not attempt
    // to reconstruct it.
    // Note that if some nodes report different values for the block length,
    // it also indicate garbage data at the end of the block. A different part
    // of the code handles this and only reconstructs the valid part of the
    // block, ie the minimum length reported by the nodes.
    Iterator<Map.Entry<Long, BlockData[]>> resultIterator
        = resultMap.entrySet().iterator();
    while (resultIterator.hasNext()) {
      Map.Entry<Long, BlockData[]> entry = resultIterator.next();
      BlockData[] blockDataArr = entry.getValue();
      for (Map.Entry<Integer, DatanodeDetails> e : sourceNodeMap.entrySet()) {
        // There should be an entry in the Array for each keyset node. If there
        // is not, this is an orphaned stripe and we should remove it from the
        // result.
        if (blockDataArr[e.getKey() - 1] == null) {
          LOG.warn("In container {} block {} does not have a putBlock entry " +
              "for index {} on datanode {} making it an orphan block / " +
              "stripe. It will not be reconstructed", containerID,
              entry.getKey(), e.getKey(), e.getValue());
          resultIterator.remove();
          break;
        }
      }
    }
    return resultMap;
  }

  /**
   * Get the effective length of each block group.
   * We can not be absolutely accurate when there is a failed stripe
   * in this block since the failed cells could be missing, and
   * we can not tell from the healthy cells whether the last stripe
   * is failed or not. But in such case we at most recover one extra
   * stripe for this block which does not confuse the client data view.
   *
   * @param blockGroup
   * @param replicaCount
   * @return
   */
  private long calcEffectiveBlockGroupLen(BlockData[] blockGroup,
      int replicaCount) {
    Preconditions.checkState(blockGroup.length == replicaCount);

    long blockGroupLen = Long.MAX_VALUE;

    for (int i = 0; i < replicaCount; i++) {
      if (blockGroup[i] == null) {
        continue;
      }

      long putBlockLen = getBlockDataLength(blockGroup[i]);
      // Use safe length is the minimum of the lengths recorded across the
      // stripe
      blockGroupLen = Math.min(putBlockLen, blockGroupLen);
    }
    return blockGroupLen == Long.MAX_VALUE ? 0 : blockGroupLen;
  }

  private long getBlockDataLength(BlockData blockData) {
    String lenStr = blockData.getMetadata()
        .get(OzoneConsts.BLOCK_GROUP_LEN_KEY_IN_PUT_BLOCK);
    // If we don't have the length, then it indicates a problem with the stripe.
    // All replica should carry the length, so if it is not there, we return 0,
    // which will cause us to set the length of the block to zero and not
    // attempt to reconstruct it.
    return (lenStr == null) ? 0 : Long.parseLong(lenStr);
  }

  public ECReconstructionMetrics getECReconstructionMetrics() {
    return this.metrics;
  }

  OptionalLong getTermOfLeaderSCM() {
    return Optional.ofNullable(context)
        .map(StateContext::getTermOfLeaderSCM)
        .orElse(OptionalLong.empty());
  }

  private static ExecutorService createThreadPoolExecutor(
      int corePoolSize, int maximumPoolSize, String threadNameFormat) {
    return new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
        60, TimeUnit.SECONDS, new SynchronousQueue<>(),
        new ThreadFactoryBuilder().setNameFormat(threadNameFormat).build(),
        new ThreadPoolExecutor.CallerRunsPolicy());
  }
}