/**
 * 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.hadoop.hbase.master.replication;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.replication.ReplicationPeerConfigUtil;
import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
import org.apache.hadoop.hbase.replication.ReplicationUtils;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.PeerModificationState;
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.UpdatePeerConfigStateData;

/**
 * The procedure for updating the config for a replication peer.
 */
@InterfaceAudience.Private
public class UpdatePeerConfigProcedure extends ModifyPeerProcedure {

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

  private ReplicationPeerConfig peerConfig;

  private ReplicationPeerConfig oldPeerConfig;

  private boolean enabled;

  public UpdatePeerConfigProcedure() {
  }

  public UpdatePeerConfigProcedure(String peerId, ReplicationPeerConfig peerConfig) {
    super(peerId);
    this.peerConfig = peerConfig;
  }

  @Override
  public PeerOperationType getPeerOperationType() {
    return PeerOperationType.UPDATE_CONFIG;
  }

  private void addToList(List<String> encodedRegionNames, String encodedRegionName,
      ReplicationQueueStorage queueStorage) throws ReplicationException {
    encodedRegionNames.add(encodedRegionName);
    if (encodedRegionNames.size() >= UPDATE_LAST_SEQ_ID_BATCH_SIZE) {
      queueStorage.removeLastSequenceIds(peerId, encodedRegionNames);
      encodedRegionNames.clear();
    }
  }

  @Override
  protected PeerModificationState nextStateAfterRefresh() {
    if (peerConfig.isSerial()) {
      if (oldPeerConfig.isSerial()) {
        // both serial, then if the ns/table-cfs configs are not changed, just go with the normal
        // way, otherwise we need to reopen the regions for the newly added tables.
        return ReplicationUtils.isNamespacesAndTableCFsEqual(peerConfig, oldPeerConfig)
          ? super.nextStateAfterRefresh()
          : PeerModificationState.SERIAL_PEER_REOPEN_REGIONS;
      } else {
        // we change the peer to serial, need to reopen all regions
        return PeerModificationState.SERIAL_PEER_REOPEN_REGIONS;
      }
    } else {
      if (oldPeerConfig.isSerial()) {
        // we remove the serial flag for peer, then we do not need to reopen all regions, but we
        // need to remove the last pushed sequence ids.
        return PeerModificationState.SERIAL_PEER_UPDATE_LAST_PUSHED_SEQ_ID;
      } else {
        // not serial for both, just go with the normal way.
        return super.nextStateAfterRefresh();
      }
    }
  }

  @Override
  protected void updateLastPushedSequenceIdForSerialPeer(MasterProcedureEnv env)
      throws IOException, ReplicationException {
    if (!oldPeerConfig.isSerial()) {
      assert peerConfig.isSerial();
      // change to serial
      setLastPushedSequenceId(env, peerConfig);
      return;
    }
    if (!peerConfig.isSerial()) {
      // remove the serial flag
      env.getReplicationPeerManager().removeAllLastPushedSeqIds(peerId);
      return;
    }
    // enter here means peerConfig and oldPeerConfig are both serial, let's find out the diffs and
    // process them
    ReplicationQueueStorage queueStorage = env.getReplicationPeerManager().getQueueStorage();
    Connection conn = env.getMasterServices().getConnection();
    Map<String, Long> lastSeqIds = new HashMap<String, Long>();
    List<String> encodedRegionNames = new ArrayList<>();
    for (TableDescriptor td : env.getMasterServices().getTableDescriptors().getAll().values()) {
      if (!td.hasGlobalReplicationScope()) {
        continue;
      }
      TableName tn = td.getTableName();
      if (ReplicationUtils.contains(oldPeerConfig, tn)) {
        if (!ReplicationUtils.contains(peerConfig, tn)) {
          // removed from peer config
          for (String encodedRegionName : MetaTableAccessor
            .getTableEncodedRegionNamesForSerialReplication(conn, tn)) {
            addToList(encodedRegionNames, encodedRegionName, queueStorage);
          }
        }
      } else if (ReplicationUtils.contains(peerConfig, tn)) {
        // newly added to peer config
        setLastPushedSequenceIdForTable(env, tn, lastSeqIds);
      }
    }
    if (!encodedRegionNames.isEmpty()) {
      queueStorage.removeLastSequenceIds(peerId, encodedRegionNames);
    }
    if (!lastSeqIds.isEmpty()) {
      queueStorage.setLastSequenceIds(peerId, lastSeqIds);
    }
  }

  @Override
  protected boolean enablePeerBeforeFinish() {
    // do not need to test reopenRegionsAfterRefresh since we can only enter here if
    // reopenRegionsAfterRefresh returns true.
    return enabled;
  }

  @Override
  protected ReplicationPeerConfig getOldPeerConfig() {
    return oldPeerConfig;
  }

  @Override
  protected ReplicationPeerConfig getNewPeerConfig() {
    return peerConfig;
  }

  @Override
  protected void prePeerModification(MasterProcedureEnv env) throws IOException {
    MasterCoprocessorHost cpHost = env.getMasterCoprocessorHost();
    if (cpHost != null) {
      cpHost.preUpdateReplicationPeerConfig(peerId, peerConfig);
    }
    ReplicationPeerDescription desc =
      env.getReplicationPeerManager().preUpdatePeerConfig(peerId, peerConfig);
    oldPeerConfig = desc.getPeerConfig();
    enabled = desc.isEnabled();
  }

  @Override
  protected void updatePeerStorage(MasterProcedureEnv env) throws ReplicationException {
    env.getReplicationPeerManager().updatePeerConfig(peerId, peerConfig);
    // if we need to jump to the special states for serial peers, then we need to disable the peer
    // first if it is not disabled yet.
    if (enabled && nextStateAfterRefresh() != super.nextStateAfterRefresh()) {
      env.getReplicationPeerManager().disablePeer(peerId);
    }
  }

  @Override
  protected void postPeerModification(MasterProcedureEnv env)
      throws IOException, ReplicationException {
    if (oldPeerConfig.isSerial() && !peerConfig.isSerial()) {
      env.getReplicationPeerManager().removeAllLastPushedSeqIds(peerId);
    }
    LOG.info("Successfully updated peer config of {} to {}", peerId, peerConfig);
    MasterCoprocessorHost cpHost = env.getMasterCoprocessorHost();
    if (cpHost != null) {
      cpHost.postUpdateReplicationPeerConfig(peerId, peerConfig);
    }
  }

  @Override
  protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException {
    super.serializeStateData(serializer);
    UpdatePeerConfigStateData.Builder builder = UpdatePeerConfigStateData.newBuilder()
      .setPeerConfig(ReplicationPeerConfigUtil.convert(peerConfig));
    if (oldPeerConfig != null) {
      builder.setOldPeerConfig(ReplicationPeerConfigUtil.convert(oldPeerConfig));
    }
    builder.setEnabled(enabled);
    serializer.serialize(builder.build());
  }

  @Override
  protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException {
    super.deserializeStateData(serializer);
    UpdatePeerConfigStateData data = serializer.deserialize(UpdatePeerConfigStateData.class);
    peerConfig = ReplicationPeerConfigUtil.convert(data.getPeerConfig());
    if (data.hasOldPeerConfig()) {
      oldPeerConfig = ReplicationPeerConfigUtil.convert(data.getOldPeerConfig());
    } else {
      oldPeerConfig = null;
    }
    enabled = data.getEnabled();
  }
}