View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.net.ConnectException;
24  import java.net.SocketTimeoutException;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.hbase.classification.InterfaceAudience;
29  import org.apache.hadoop.conf.Configuration;
30  import org.apache.hadoop.fs.FileSystem;
31  import org.apache.hadoop.fs.Path;
32  import org.apache.hadoop.hbase.NotServingRegionException;
33  import org.apache.hadoop.hbase.Server;
34  import org.apache.hadoop.hbase.client.HConnectionManager;
35  import org.apache.hadoop.hbase.client.RetriesExhaustedException;
36  import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
37  import org.apache.hadoop.hbase.coordination.SplitLogWorkerCoordination;
38  import org.apache.hadoop.hbase.master.SplitLogManager;
39  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
40  import org.apache.hadoop.hbase.regionserver.wal.HLogSplitter;
41  import org.apache.hadoop.hbase.util.CancelableProgressable;
42  import org.apache.hadoop.hbase.util.ExceptionUtil;
43  import org.apache.hadoop.hbase.util.FSUtils;
44  
45  import com.google.common.annotations.VisibleForTesting;
46  
47  /**
48   * This worker is spawned in every regionserver, including master. The Worker waits for log
49   * splitting tasks to be put up by the {@link SplitLogManager} running in the master and races with
50   * other workers in other serves to acquire those tasks. The coordination is done via coordination
51   * engine.
52   * <p>
53   * If a worker has successfully moved the task from state UNASSIGNED to OWNED then it owns the task.
54   * It keeps heart beating the manager by periodically moving the task from UNASSIGNED to OWNED
55   * state. On success it moves the task to TASK_DONE. On unrecoverable error it moves task state to
56   * ERR. If it cannot continue but wants the master to retry the task then it moves the task state to
57   * RESIGNED.
58   * <p>
59   * The manager can take a task away from a worker by moving the task from OWNED to UNASSIGNED. In
60   * the absence of a global lock there is a unavoidable race here - a worker might have just finished
61   * its task when it is stripped of its ownership. Here we rely on the idempotency of the log
62   * splitting task for correctness
63   */
64  @InterfaceAudience.Private
65  public class SplitLogWorker implements Runnable {
66  
67    private static final Log LOG = LogFactory.getLog(SplitLogWorker.class);
68  
69    Thread worker;
70    // thread pool which executes recovery work
71    private SplitLogWorkerCoordination coordination;
72    private Configuration conf;
73    private RegionServerServices server;
74    public SplitLogWorker(Server hserver, Configuration conf, RegionServerServices server,
75        TaskExecutor splitTaskExecutor) {
76      this.server = server;
77      this.conf = conf;
78      this.coordination =
79          ((BaseCoordinatedStateManager) hserver.getCoordinatedStateManager())
80              .getSplitLogWorkerCoordination();
81      this.server = server;
82      coordination.init(server, conf, splitTaskExecutor, this);
83    }
84  
85    public SplitLogWorker(final Server hserver, final Configuration conf,
86        final RegionServerServices server, final LastSequenceId sequenceIdChecker) {
87      this(server, conf, server, new TaskExecutor() {
88        @Override
89        public Status exec(String filename, RecoveryMode mode, CancelableProgressable p) {
90          Path rootdir;
91          FileSystem fs;
92          try {
93            rootdir = FSUtils.getRootDir(conf);
94            fs = rootdir.getFileSystem(conf);
95          } catch (IOException e) {
96            LOG.warn("could not find root dir or fs", e);
97            return Status.RESIGNED;
98          }
99          // TODO have to correctly figure out when log splitting has been
100         // interrupted or has encountered a transient error and when it has
101         // encountered a bad non-retry-able persistent error.
102         try {
103           if (!HLogSplitter.splitLogFile(rootdir, fs.getFileStatus(new Path(rootdir, filename)),
104             fs, conf, p, sequenceIdChecker, server.getCoordinatedStateManager(), mode)) {
105             return Status.PREEMPTED;
106           }
107         } catch (InterruptedIOException iioe) {
108           LOG.warn("log splitting of " + filename + " interrupted, resigning", iioe);
109           return Status.RESIGNED;
110         } catch (IOException e) {
111           Throwable cause = e.getCause();
112           if (e instanceof RetriesExhaustedException && (cause instanceof NotServingRegionException
113                   || cause instanceof ConnectException
114                   || cause instanceof SocketTimeoutException)) {
115             LOG.warn("log replaying of " + filename + " can't connect to the target regionserver, "
116             		+ "resigning", e);
117             return Status.RESIGNED;
118           } else if (cause instanceof InterruptedException) {
119             LOG.warn("log splitting of " + filename + " interrupted, resigning", e);
120             return Status.RESIGNED;
121           }
122           LOG.warn("log splitting of " + filename + " failed, returning error", e);
123           return Status.ERR;
124         }
125         return Status.DONE;
126       }
127     });
128   }
129 
130   @Override
131   public void run() {
132     try {
133       LOG.info("SplitLogWorker " + server.getServerName() + " starting");
134       coordination.registerListener();
135       // pre-initialize a new connection for splitlogworker configuration
136       HConnectionManager.getConnection(conf);
137 
138       // wait for Coordination Engine is ready
139       boolean res = false;
140       while (!res && !coordination.isStop()) {
141         res = coordination.isReady();
142       }
143       if (!coordination.isStop()) {
144         coordination.taskLoop();
145       }
146     } catch (Throwable t) {
147       if (ExceptionUtil.isInterrupt(t)) {
148         LOG.info("SplitLogWorker interrupted. Exiting. " + (coordination.isStop() ? "" :
149             " (ERROR: exitWorker is not set, exiting anyway)"));
150       } else {
151         // only a logical error can cause here. Printing it out
152         // to make debugging easier
153         LOG.error("unexpected error ", t);
154       }
155     } finally {
156       coordination.removeListener();
157       LOG.info("SplitLogWorker " + server.getServerName() + " exiting");
158     }
159   }
160   /**
161    * If the worker is doing a task i.e. splitting a log file then stop the task.
162    * It doesn't exit the worker thread.
163    */
164   public void stopTask() {
165     LOG.info("Sending interrupt to stop the worker thread");
166     worker.interrupt(); // TODO interrupt often gets swallowed, do what else?
167   }
168 
169   /**
170    * start the SplitLogWorker thread
171    */
172   public void start() {
173     worker = new Thread(null, this, "SplitLogWorker-" + server.getServerName().toShortString());
174     worker.start();
175   }
176 
177   /**
178    * stop the SplitLogWorker thread
179    */
180   public void stop() {
181     coordination.stopProcessingTasks();
182     stopTask();
183   }
184 
185   /**
186    * Objects implementing this interface actually do the task that has been
187    * acquired by a {@link SplitLogWorker}. Since there isn't a water-tight
188    * guarantee that two workers will not be executing the same task therefore it
189    * is better to have workers prepare the task and then have the
190    * {@link SplitLogManager} commit the work in SplitLogManager.TaskFinisher
191    */
192   public interface TaskExecutor {
193     enum Status {
194       DONE(),
195       ERR(),
196       RESIGNED(),
197       PREEMPTED()
198     }
199     Status exec(String name, RecoveryMode mode, CancelableProgressable p);
200   }
201 
202   /**
203    * Returns the number of tasks processed by coordination.
204    * This method is used by tests only
205    */
206   @VisibleForTesting
207   public int getTaskReadySeq() {
208     return coordination.getTaskReadySeq();
209   }
210 }