content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.asterix.replication.sync;
  
  import static org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation.DELETE;
  import static org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation.REPLICATE;
  
  import java.io.IOException;
  
  import org.apache.asterix.common.api.INcApplicationContext;
  import org.apache.asterix.common.ioopcallbacks.LSMIOOperationCallback;
  import org.apache.asterix.common.utils.StoragePathUtil;
  import org.apache.asterix.replication.api.PartitionReplica;
  import org.apache.asterix.replication.messaging.ComponentMaskTask;
  import org.apache.asterix.replication.messaging.DropIndexTask;
  import org.apache.asterix.replication.messaging.MarkComponentValidTask;
  import org.apache.asterix.replication.messaging.ReplicationProtocol;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.api.replication.IReplicationJob;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexReplicationJob;
  import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
  import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  
  public class IndexSynchronizer {
  
      private static final Logger LOGGER = LogManager.getLogger();
      public static final long MERGE_LSN = -1;
      public static final long BULKLOAD_LSN = -2;
      private final IReplicationJob job;
      private final INcApplicationContext appCtx;
  
      public IndexSynchronizer(IReplicationJob job, INcApplicationContext appCtx) {
          this.job = job;
          this.appCtx = appCtx;
      }
  
      public void sync(PartitionReplica replica) throws IOException {
          switch (job.getJobType()) {
              case LSM_COMPONENT:
                  syncComponent(replica);
                  break;
              case METADATA:
                  syncMetadata(replica);
                  break;
              default:
                  throw new IllegalStateException("unrecognized job type: " + job.getJobType().name());
          }
      }
  
      private void syncComponent(PartitionReplica replica) throws IOException {
          if (job.getOperation() == REPLICATE) {
              replicateComponent(replica);
          } else if (job.getOperation() == DELETE) {
              deleteComponent(replica);
          }
      }
  
      private void syncMetadata(PartitionReplica replica) throws IOException {
          if (job.getOperation() == REPLICATE) {
              replicateIndexMetadata(replica);
          } else if (job.getOperation() == DELETE) {
              deleteIndexMetadata(replica);
          }
      }
  
      private void replicateComponent(PartitionReplica replica) throws IOException {
          // send component header
          final String anyFile = job.getAnyFile();
          final String indexFile = StoragePathUtil.getFileRelativePath(anyFile);
          final ComponentMaskTask maskTask = new ComponentMaskTask(indexFile);
          ReplicationProtocol.sendTo(replica, maskTask);
          ReplicationProtocol.waitForAck(replica);
          // send component files
          final FileSynchronizer fileSynchronizer = new FileSynchronizer(appCtx, replica);
          job.getJobFiles().stream().map(StoragePathUtil::getFileRelativePath).forEach(fileSynchronizer::replicate);
          // send mark component valid
          String masterNode = appCtx.getReplicaManager().isPartitionOrigin(replica.getIdentifier().getPartition())
                  ? appCtx.getServiceContext().getNodeId() : null;
          MarkComponentValidTask markValidTask = new MarkComponentValidTask(indexFile, getReplicatedComponentLsn(),
                  getReplicatedComponentId(), masterNode);
          ReplicationProtocol.sendTo(replica, markValidTask);
          ReplicationProtocol.waitForAck(replica);
          LOGGER.debug("Replicated component ({}) to replica {}", indexFile, replica);
      }
  
      private void deleteComponent(PartitionReplica replica) {
          FileSynchronizer fileSynchronizer = new FileSynchronizer(appCtx, replica);
          job.getJobFiles().stream().map(StoragePathUtil::getFileRelativePath).forEach(fileSynchronizer::delete);
      }
  
      private void replicateIndexMetadata(PartitionReplica replica) {
          // send the index metadata file
          final FileSynchronizer fileSynchronizer = new FileSynchronizer(appCtx, replica);
          job.getJobFiles().stream().map(StoragePathUtil::getFileRelativePath)
                  .forEach(file -> fileSynchronizer.replicate(file, true));
      }
  
      private void deleteIndexMetadata(PartitionReplica replica) throws IOException {
          final String file = StoragePathUtil.getFileRelativePath(job.getAnyFile());
          final DropIndexTask task = new DropIndexTask(file);
          ReplicationProtocol.sendTo(replica, task);
          ReplicationProtocol.waitForAck(replica);
      }
  
      private long getReplicatedComponentLsn() throws HyracksDataException {
          final ILSMIndexReplicationJob indexReplJob = (ILSMIndexReplicationJob) job;
          if (indexReplJob.getLSMOpType() == LSMOperationType.MERGE) {
              return MERGE_LSN;
          } else if (indexReplJob.getLSMOpType() == LSMOperationType.LOAD) {
              return BULKLOAD_LSN;
          }
  
          if (indexReplJob.getLSMOpType() != LSMOperationType.FLUSH) {
              return LSMIOOperationCallback.INVALID_LSN;
          }
          final ILSMIndex lsmIndex = indexReplJob.getLSMIndex();
          final ILSMIndexOperationContext ctx = indexReplJob.getLSMIndexOperationContext();
          return ((LSMIOOperationCallback) lsmIndex.getIOOperationCallback())
                  .getComponentLSN(ctx.getComponentsToBeReplicated());
      }
  
      private long getReplicatedComponentId() throws HyracksDataException {
          final ILSMIndexReplicationJob indexReplJob = (ILSMIndexReplicationJob) job;
          if (indexReplJob.getLSMOpType() != LSMOperationType.FLUSH
                  && indexReplJob.getLSMOpType() != LSMOperationType.LOAD) {
              return -1L;
          }
          final ILSMIndexOperationContext ctx = indexReplJob.getLSMIndexOperationContext();
          LSMComponentId id = (LSMComponentId) ctx.getComponentsToBeReplicated().get(0).getId();
          return id.getMaxId();
      }
  }
  