001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.master.replication; 019 020import java.io.IOException; 021import java.util.ArrayList; 022import java.util.List; 023import java.util.Optional; 024import org.apache.hadoop.hbase.ServerName; 025import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; 026import org.apache.hadoop.hbase.master.procedure.PeerProcedureInterface; 027import org.apache.hadoop.hbase.master.procedure.RSProcedureDispatcher.ServerOperation; 028import org.apache.hadoop.hbase.master.procedure.ServerRemoteProcedure; 029import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer; 030import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteOperation; 031import org.apache.hadoop.hbase.replication.regionserver.ReplaySyncReplicationWALCallable; 032import org.apache.yetus.audience.InterfaceAudience; 033import org.slf4j.Logger; 034import org.slf4j.LoggerFactory; 035 036import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 037import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.ReplaySyncReplicationWALParameter; 038import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.SyncReplicationReplayWALRemoteStateData; 039 040/** 041 * A remote procedure which is used to send replaying remote wal work to region server. 042 */ 043@InterfaceAudience.Private 044public class SyncReplicationReplayWALRemoteProcedure extends ServerRemoteProcedure 045 implements PeerProcedureInterface { 046 047 private static final Logger LOG = 048 LoggerFactory.getLogger(SyncReplicationReplayWALRemoteProcedure.class); 049 050 private String peerId; 051 052 private List<String> wals; 053 054 public SyncReplicationReplayWALRemoteProcedure() { 055 } 056 057 public SyncReplicationReplayWALRemoteProcedure(String peerId, List<String> wals, 058 ServerName targetServer) { 059 this.peerId = peerId; 060 this.wals = wals; 061 this.targetServer = targetServer; 062 } 063 064 @Override 065 public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerName remote) { 066 ReplaySyncReplicationWALParameter.Builder builder = 067 ReplaySyncReplicationWALParameter.newBuilder(); 068 builder.setPeerId(peerId); 069 wals.stream().forEach(builder::addWal); 070 return Optional.of(new ServerOperation(this, getProcId(), 071 ReplaySyncReplicationWALCallable.class, builder.build().toByteArray())); 072 } 073 074 protected void complete(MasterProcedureEnv env, Throwable error) { 075 if (error != null) { 076 LOG.warn("Replay wals {} on {} failed for peer id={}", wals, targetServer, peerId, error); 077 this.succ = false; 078 } else { 079 truncateWALs(env); 080 LOG.info("Replay wals {} on {} succeed for peer id={}", wals, targetServer, peerId); 081 this.succ = true; 082 } 083 } 084 085 /** 086 * Only truncate wals one by one when task succeed. The parent procedure will check the first wal 087 * length to know whether this task succeed. 088 */ 089 private void truncateWALs(MasterProcedureEnv env) { 090 String firstWal = wals.get(0); 091 try { 092 env.getMasterServices().getSyncReplicationReplayWALManager().finishReplayWAL(firstWal); 093 } catch (IOException e) { 094 // As it is idempotent to rerun this task. Just ignore this exception and return. 095 LOG.warn("Failed to truncate wal {} for peer id={}", firstWal, peerId, e); 096 return; 097 } 098 for (int i = 1; i < wals.size(); i++) { 099 String wal = wals.get(i); 100 try { 101 env.getMasterServices().getSyncReplicationReplayWALManager().finishReplayWAL(wal); 102 } catch (IOException e1) { 103 try { 104 // retry 105 env.getMasterServices().getSyncReplicationReplayWALManager().finishReplayWAL(wal); 106 } catch (IOException e2) { 107 // As the parent procedure only check the first wal length. Just ignore this exception. 108 LOG.warn("Failed to truncate wal {} for peer id={}", wal, peerId, e2); 109 } 110 } 111 } 112 } 113 114 @Override 115 protected void rollback(MasterProcedureEnv env) throws IOException, InterruptedException { 116 throw new UnsupportedOperationException(); 117 } 118 119 @Override 120 protected boolean abort(MasterProcedureEnv env) { 121 return false; 122 } 123 124 @Override 125 protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException { 126 SyncReplicationReplayWALRemoteStateData.Builder builder = 127 SyncReplicationReplayWALRemoteStateData.newBuilder().setPeerId(peerId) 128 .setTargetServer(ProtobufUtil.toServerName(targetServer)); 129 wals.stream().forEach(builder::addWal); 130 serializer.serialize(builder.build()); 131 } 132 133 @Override 134 protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException { 135 SyncReplicationReplayWALRemoteStateData data = 136 serializer.deserialize(SyncReplicationReplayWALRemoteStateData.class); 137 peerId = data.getPeerId(); 138 wals = new ArrayList<>(); 139 data.getWalList().forEach(wals::add); 140 targetServer = ProtobufUtil.toServerName(data.getTargetServer()); 141 } 142 143 @Override 144 public String getPeerId() { 145 return peerId; 146 } 147 148 @Override 149 public PeerOperationType getPeerOperationType() { 150 return PeerOperationType.SYNC_REPLICATION_REPLAY_WAL_REMOTE; 151 } 152}