001/** 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one 004 * or more contributor license agreements. See the NOTICE file 005 * distributed with this work for additional information 006 * regarding copyright ownership. The ASF licenses this file 007 * to you under the Apache License, Version 2.0 (the 008 * "License"); you may not use this file except in compliance 009 * with the License. You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 */ 019package org.apache.hadoop.hbase.regionserver.handler; 020 021import java.io.IOException; 022import java.util.concurrent.atomic.AtomicInteger; 023 024import org.apache.yetus.audience.InterfaceAudience; 025import org.slf4j.Logger; 026import org.slf4j.LoggerFactory; 027import org.apache.hadoop.hbase.Server; 028import org.apache.hadoop.hbase.ServerName; 029import org.apache.hadoop.hbase.SplitLogCounters; 030import org.apache.hadoop.hbase.SplitLogTask; 031import org.apache.hadoop.hbase.coordination.SplitLogWorkerCoordination; 032import org.apache.hadoop.hbase.executor.EventHandler; 033import org.apache.hadoop.hbase.executor.EventType; 034import org.apache.hadoop.hbase.regionserver.SplitLogWorker.TaskExecutor; 035import org.apache.hadoop.hbase.regionserver.SplitLogWorker.TaskExecutor.Status; 036import org.apache.hadoop.hbase.util.CancelableProgressable; 037 038/** 039 * Handles log splitting a wal 040 * Used by the zk-based distributed log splitting. Created by ZKSplitLogWorkerCoordination. 041 * @deprecated since 2.4.0 and in 3.0.0, to be removed in 4.0.0, replaced by procedure-based 042 * distributed WAL splitter, see SplitWALManager 043 */ 044@Deprecated 045@InterfaceAudience.Private 046public class WALSplitterHandler extends EventHandler { 047 private static final Logger LOG = LoggerFactory.getLogger(WALSplitterHandler.class); 048 private final ServerName serverName; 049 private final CancelableProgressable reporter; 050 private final AtomicInteger inProgressTasks; 051 private final TaskExecutor splitTaskExecutor; 052 private final SplitLogWorkerCoordination.SplitTaskDetails splitTaskDetails; 053 private final SplitLogWorkerCoordination coordination; 054 055 056 public WALSplitterHandler(final Server server, SplitLogWorkerCoordination coordination, 057 SplitLogWorkerCoordination.SplitTaskDetails splitDetails, CancelableProgressable reporter, 058 AtomicInteger inProgressTasks, TaskExecutor splitTaskExecutor) { 059 super(server, EventType.RS_LOG_REPLAY); 060 this.splitTaskDetails = splitDetails; 061 this.coordination = coordination; 062 this.reporter = reporter; 063 this.inProgressTasks = inProgressTasks; 064 this.inProgressTasks.incrementAndGet(); 065 this.serverName = server.getServerName(); 066 this.splitTaskExecutor = splitTaskExecutor; 067 } 068 069 @Override 070 public void process() throws IOException { 071 long startTime = System.currentTimeMillis(); 072 Status status = null; 073 try { 074 status = this.splitTaskExecutor.exec(splitTaskDetails.getWALFile(), reporter); 075 switch (status) { 076 case DONE: 077 coordination.endTask(new SplitLogTask.Done(this.serverName), 078 SplitLogCounters.tot_wkr_task_done, splitTaskDetails); 079 break; 080 case PREEMPTED: 081 SplitLogCounters.tot_wkr_preempt_task.increment(); 082 LOG.warn("task execution preempted " + splitTaskDetails.getWALFile()); 083 break; 084 case ERR: 085 if (server != null && !server.isStopped()) { 086 coordination.endTask(new SplitLogTask.Err(this.serverName), 087 SplitLogCounters.tot_wkr_task_err, splitTaskDetails); 088 break; 089 } 090 // if the RS is exiting then there is probably a tons of stuff 091 // that can go wrong. Resign instead of signaling error. 092 //$FALL-THROUGH$ 093 case RESIGNED: 094 if (server != null && server.isStopped()) { 095 LOG.info("task execution interrupted because worker is exiting " 096 + splitTaskDetails.toString()); 097 } 098 coordination.endTask(new SplitLogTask.Resigned(this.serverName), 099 SplitLogCounters.tot_wkr_task_resigned, splitTaskDetails); 100 break; 101 } 102 } finally { 103 LOG.info("Worker " + serverName + " done with task " + splitTaskDetails.toString() + " in " 104 + (System.currentTimeMillis() - startTime) + "ms. Status = " + status); 105 this.inProgressTasks.decrementAndGet(); 106 } 107 } 108}