View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.procedure2.store.wal;
20  
21  import java.io.IOException;
22  import java.util.Iterator;
23  import java.util.Map;
24  import java.util.HashMap;
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.hbase.classification.InterfaceStability;
30  import org.apache.hadoop.fs.FSDataInputStream;
31  import org.apache.hadoop.hbase.procedure2.Procedure;
32  import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreTracker;
33  import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos;
34  import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureWALEntry;
35  
36  /**
37   * Helper class that loads the procedures stored in a WAL
38   */
39  @InterfaceAudience.Private
40  @InterfaceStability.Evolving
41  public class ProcedureWALFormatReader {
42    private static final Log LOG = LogFactory.getLog(ProcedureWALFormatReader.class);
43  
44    private final ProcedureStoreTracker tracker;
45    //private final long compactionLogId;
46  
47    private final Map<Long, Procedure> procedures = new HashMap<Long, Procedure>();
48    private final Map<Long, ProcedureProtos.Procedure> localProcedures =
49      new HashMap<Long, ProcedureProtos.Procedure>();
50  
51    private long maxProcId = 0;
52  
53    public ProcedureWALFormatReader(final ProcedureStoreTracker tracker) {
54      this.tracker = tracker;
55    }
56  
57    public void read(ProcedureWALFile log, ProcedureWALFormat.Loader loader) throws IOException {
58      FSDataInputStream stream = log.getStream();
59      try {
60        boolean hasMore = true;
61        while (hasMore) {
62          ProcedureWALEntry entry = ProcedureWALFormat.readEntry(stream);
63          if (entry == null) {
64            LOG.warn("nothing left to decode. exiting with missing EOF");
65            hasMore = false;
66            break;
67          }
68          switch (entry.getType()) {
69            case INIT:
70              readInitEntry(entry);
71              break;
72            case INSERT:
73              readInsertEntry(entry);
74              break;
75            case UPDATE:
76            case COMPACT:
77              readUpdateEntry(entry);
78              break;
79            case DELETE:
80              readDeleteEntry(entry);
81              break;
82            case EOF:
83              hasMore = false;
84              break;
85            default:
86              throw new CorruptedWALProcedureStoreException("Invalid entry: " + entry);
87          }
88        }
89      } catch (IOException e) {
90        LOG.error("got an exception while reading the procedure WAL: " + log, e);
91        loader.markCorruptedWAL(log, e);
92      }
93  
94      if (localProcedures.isEmpty()) {
95        LOG.info("No active entry found in state log " + log + ". removing it");
96        loader.removeLog(log);
97      } else {
98        Iterator<Map.Entry<Long, ProcedureProtos.Procedure>> itd =
99          localProcedures.entrySet().iterator();
100       long minProcId = Long.MAX_VALUE;
101       long maxProcId = Long.MIN_VALUE;
102       while (itd.hasNext()) {
103         Map.Entry<Long, ProcedureProtos.Procedure> entry = itd.next();
104         itd.remove();
105 
106         long procId = entry.getKey();
107         minProcId = Math.min(minProcId, procId);
108         maxProcId = Math.max(maxProcId, procId);
109 
110         // Deserialize the procedure
111         Procedure proc = Procedure.convert(entry.getValue());
112         procedures.put(procId, proc);
113       }
114 
115       // TODO: Some procedure may be already runnables (see readInitEntry())
116       //       (we can also check the "update map" in the log trackers)
117       log.setProcIds(minProcId, maxProcId);
118     }
119   }
120 
121   public Iterator<Procedure> getProcedures() {
122     return procedures.values().iterator();
123   }
124 
125   private void loadEntries(final ProcedureWALEntry entry) {
126     for (ProcedureProtos.Procedure proc: entry.getProcedureList()) {
127       maxProcId = Math.max(maxProcId, proc.getProcId());
128       if (isRequired(proc.getProcId())) {
129         if (LOG.isTraceEnabled()) {
130           LOG.trace("read " + entry.getType() + " entry " + proc.getProcId());
131         }
132         localProcedures.put(proc.getProcId(), proc);
133         tracker.setDeleted(proc.getProcId(), false);
134       }
135     }
136   }
137 
138   private void readInitEntry(final ProcedureWALEntry entry)
139       throws IOException {
140     assert entry.getProcedureCount() == 1 : "Expected only one procedure";
141     // TODO: Make it runnable, before reading other files
142     loadEntries(entry);
143   }
144 
145   private void readInsertEntry(final ProcedureWALEntry entry) throws IOException {
146     assert entry.getProcedureCount() >= 1 : "Expected one or more procedures";
147     loadEntries(entry);
148   }
149 
150   private void readUpdateEntry(final ProcedureWALEntry entry) throws IOException {
151     assert entry.getProcedureCount() == 1 : "Expected only one procedure";
152     loadEntries(entry);
153   }
154 
155   private void readDeleteEntry(final ProcedureWALEntry entry) throws IOException {
156     assert entry.getProcedureCount() == 0 : "Expected no procedures";
157     assert entry.hasProcId() : "expected ProcID";
158     if (LOG.isTraceEnabled()) {
159       LOG.trace("read delete entry " + entry.getProcId());
160     }
161     maxProcId = Math.max(maxProcId, entry.getProcId());
162     localProcedures.remove(entry.getProcId());
163     tracker.setDeleted(entry.getProcId(), true);
164   }
165 
166   private boolean isDeleted(final long procId) {
167     return tracker.isDeleted(procId) == ProcedureStoreTracker.DeleteState.YES;
168   }
169 
170   private boolean isRequired(final long procId) {
171     return !isDeleted(procId) && !procedures.containsKey(procId);
172   }
173 }