1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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
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
111 Procedure proc = Procedure.convert(entry.getValue());
112 procedures.put(procId, proc);
113 }
114
115
116
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
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 }