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.regionserver; 019 020import java.io.FileNotFoundException; 021import java.io.IOException; 022import java.io.InterruptedIOException; 023import java.net.ConnectException; 024import java.net.SocketTimeoutException; 025import org.apache.hadoop.conf.Configuration; 026import org.apache.hadoop.fs.FileSystem; 027import org.apache.hadoop.fs.Path; 028import org.apache.hadoop.hbase.NotServingRegionException; 029import org.apache.hadoop.hbase.Server; 030import org.apache.hadoop.hbase.client.RetriesExhaustedException; 031import org.apache.hadoop.hbase.coordination.SplitLogWorkerCoordination; 032import org.apache.hadoop.hbase.regionserver.SplitLogWorker.TaskExecutor.Status; 033import org.apache.hadoop.hbase.util.CancelableProgressable; 034import org.apache.hadoop.hbase.util.CommonFSUtils; 035import org.apache.hadoop.hbase.util.ExceptionUtil; 036import org.apache.hadoop.hbase.wal.WALFactory; 037import org.apache.hadoop.hbase.wal.WALSplitter; 038import org.apache.yetus.audience.InterfaceAudience; 039import org.slf4j.Logger; 040import org.slf4j.LoggerFactory; 041 042/** 043 * This worker is spawned in every regionserver, including master. The Worker waits for log 044 * splitting tasks to be put up by the {@link org.apache.hadoop.hbase.master.SplitLogManager} 045 * running in the master and races with other workers in other serves to acquire those tasks. 046 * The coordination is done via coordination engine. 047 * <p> 048 * If a worker has successfully moved the task from state UNASSIGNED to OWNED then it owns the task. 049 * It keeps heart beating the manager by periodically moving the task from UNASSIGNED to OWNED 050 * state. On success it moves the task to TASK_DONE. On unrecoverable error it moves task state to 051 * ERR. If it cannot continue but wants the master to retry the task then it moves the task state to 052 * RESIGNED. 053 * <p> 054 * The manager can take a task away from a worker by moving the task from OWNED to UNASSIGNED. In 055 * the absence of a global lock there is a unavoidable race here - a worker might have just finished 056 * its task when it is stripped of its ownership. Here we rely on the idempotency of the log 057 * splitting task for correctness 058 * @deprecated since 2.4.0 and in 3.0.0, to be removed in 4.0.0, replaced by procedure-based 059 * distributed WAL splitter, see SplitWALRemoteProcedure 060 */ 061@Deprecated 062@InterfaceAudience.Private 063public class SplitLogWorker implements Runnable { 064 065 private static final Logger LOG = LoggerFactory.getLogger(SplitLogWorker.class); 066 067 Thread worker; 068 // thread pool which executes recovery work 069 private final SplitLogWorkerCoordination coordination; 070 private final Configuration conf; 071 private final RegionServerServices server; 072 073 public SplitLogWorker(Server hserver, Configuration conf, RegionServerServices server, 074 TaskExecutor splitTaskExecutor) { 075 this.server = server; 076 // Unused. 077 this.conf = conf; 078 this.coordination = hserver.getCoordinatedStateManager().getSplitLogWorkerCoordination(); 079 coordination.init(server, conf, splitTaskExecutor, this); 080 } 081 082 public SplitLogWorker(Configuration conf, RegionServerServices server, 083 LastSequenceId sequenceIdChecker, WALFactory factory) { 084 this(server, conf, server, (f, p) -> splitLog(f, p, conf, server, sequenceIdChecker, factory)); 085 } 086 087 /** 088 * @return Result either DONE, RESIGNED, or ERR. 089 */ 090 static Status splitLog(String filename, CancelableProgressable p, Configuration conf, 091 RegionServerServices server, LastSequenceId sequenceIdChecker, WALFactory factory) { 092 Path walDir; 093 FileSystem fs; 094 try { 095 walDir = CommonFSUtils.getWALRootDir(conf); 096 fs = walDir.getFileSystem(conf); 097 } catch (IOException e) { 098 LOG.warn("Resigning, could not find root dir or fs", e); 099 return Status.RESIGNED; 100 } 101 // TODO have to correctly figure out when log splitting has been 102 // interrupted or has encountered a transient error and when it has 103 // encountered a bad non-retry-able persistent error. 104 try { 105 SplitLogWorkerCoordination splitLogWorkerCoordination = 106 server.getCoordinatedStateManager() == null ? null 107 : server.getCoordinatedStateManager().getSplitLogWorkerCoordination(); 108 if (!WALSplitter.splitLogFile(walDir, fs.getFileStatus(new Path(walDir, filename)), fs, conf, 109 p, sequenceIdChecker, splitLogWorkerCoordination, factory, server)) { 110 return Status.PREEMPTED; 111 } 112 } catch (InterruptedIOException iioe) { 113 LOG.warn("Resigning, interrupted splitting WAL {}", filename, iioe); 114 return Status.RESIGNED; 115 } catch (IOException e) { 116 if (e instanceof FileNotFoundException) { 117 // A wal file may not exist anymore. Nothing can be recovered so move on 118 LOG.warn("Done, WAL {} does not exist anymore", filename, e); 119 return Status.DONE; 120 } 121 Throwable cause = e.getCause(); 122 if (e instanceof RetriesExhaustedException && (cause instanceof NotServingRegionException 123 || cause instanceof ConnectException || cause instanceof SocketTimeoutException)) { 124 LOG.warn("Resigning, can't connect to target regionserver splitting WAL {}", filename, e); 125 return Status.RESIGNED; 126 } else if (cause instanceof InterruptedException) { 127 LOG.warn("Resigning, interrupted splitting WAL {}", filename, e); 128 return Status.RESIGNED; 129 } 130 LOG.warn("Error splitting WAL {}", filename, e); 131 return Status.ERR; 132 } 133 LOG.debug("Done splitting WAL {}", filename); 134 return Status.DONE; 135 } 136 137 138 @Override 139 public void run() { 140 try { 141 LOG.info("SplitLogWorker " + server.getServerName() + " starting"); 142 coordination.registerListener(); 143 // wait for Coordination Engine is ready 144 boolean res = false; 145 while (!res && !coordination.isStop()) { 146 res = coordination.isReady(); 147 } 148 if (!coordination.isStop()) { 149 coordination.taskLoop(); 150 } 151 } catch (Throwable t) { 152 if (ExceptionUtil.isInterrupt(t)) { 153 LOG.info("SplitLogWorker interrupted. Exiting. " + (coordination.isStop() ? "" : 154 " (ERROR: exitWorker is not set, exiting anyway)")); 155 } else { 156 // only a logical error can cause here. Printing it out 157 // to make debugging easier 158 LOG.error("unexpected error ", t); 159 } 160 } finally { 161 coordination.removeListener(); 162 LOG.info("SplitLogWorker " + server.getServerName() + " exiting"); 163 } 164 } 165 166 /** 167 * If the worker is doing a task i.e. splitting a log file then stop the task. 168 * It doesn't exit the worker thread. 169 */ 170 public void stopTask() { 171 LOG.info("Sending interrupt to stop the worker thread"); 172 worker.interrupt(); // TODO interrupt often gets swallowed, do what else? 173 } 174 175 /** 176 * start the SplitLogWorker thread 177 */ 178 public void start() { 179 worker = new Thread(null, this, "SplitLogWorker-" + server.getServerName().toShortString()); 180 worker.start(); 181 } 182 183 /** 184 * stop the SplitLogWorker thread 185 */ 186 public void stop() { 187 coordination.stopProcessingTasks(); 188 stopTask(); 189 } 190 191 /** 192 * Objects implementing this interface actually do the task that has been 193 * acquired by a {@link SplitLogWorker}. Since there isn't a water-tight 194 * guarantee that two workers will not be executing the same task therefore it 195 * is better to have workers prepare the task and then have the 196 * {@link org.apache.hadoop.hbase.master.SplitLogManager} commit the work in 197 * SplitLogManager.TaskFinisher 198 */ 199 public interface TaskExecutor { 200 enum Status { 201 DONE(), 202 ERR(), 203 RESIGNED(), 204 PREEMPTED() 205 } 206 Status exec(String name, CancelableProgressable p); 207 } 208 209 /** 210 * Returns the number of tasks processed by coordination. 211 * This method is used by tests only 212 */ 213 public int getTaskReadySeq() { 214 return coordination.getTaskReadySeq(); 215 } 216}