1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import static org.apache.hadoop.hbase.regionserver.StripeStoreFileManager.OPEN_KEY;
22
23 import java.io.IOException;
24 import java.util.ArrayList;
25 import java.util.List;
26
27 import com.google.common.annotations.VisibleForTesting;
28
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31 import org.apache.hadoop.conf.Configuration;
32 import org.apache.hadoop.fs.Path;
33 import org.apache.hadoop.hbase.KeyValue.KVComparator;
34 import org.apache.hadoop.hbase.classification.InterfaceAudience;
35 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
36 import org.apache.hadoop.hbase.regionserver.StoreFile.Writer;
37 import org.apache.hadoop.hbase.regionserver.compactions.StripeCompactionPolicy;
38 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
39
40
41
42
43
44 @InterfaceAudience.Private
45 public class StripeStoreFlusher extends StoreFlusher {
46 private static final Log LOG = LogFactory.getLog(StripeStoreFlusher.class);
47 private final Object flushLock = new Object();
48 private final StripeCompactionPolicy policy;
49 private final StripeCompactionPolicy.StripeInformationProvider stripes;
50
51 public StripeStoreFlusher(Configuration conf, Store store,
52 StripeCompactionPolicy policy, StripeStoreFileManager stripes) {
53 super(conf, store);
54 this.policy = policy;
55 this.stripes = stripes;
56 }
57
58 @Override
59 public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushSeqNum,
60 MonitoredTask status, ThroughputController throughputController) throws IOException {
61 List<Path> result = new ArrayList<Path>();
62 int cellsCount = snapshot.getCellsCount();
63 if (cellsCount == 0) return result;
64
65 long smallestReadPoint = store.getSmallestReadPoint();
66 InternalScanner scanner = createScanner(snapshot.getScanner(), smallestReadPoint);
67 if (scanner == null) {
68 return result;
69 }
70
71
72 StripeFlushRequest req = this.policy.selectFlush(store.getComparator(), this.stripes,
73 cellsCount);
74
75 boolean success = false;
76 StripeMultiFileWriter mw = null;
77 try {
78 mw = req.createWriter();
79 StripeMultiFileWriter.WriterFactory factory = createWriterFactory(
80 snapshot.getTimeRangeTracker(), cellsCount);
81 StoreScanner storeScanner = (scanner instanceof StoreScanner) ? (StoreScanner)scanner : null;
82 mw.init(storeScanner, factory);
83
84 synchronized (flushLock) {
85 performFlush(scanner, mw, smallestReadPoint, throughputController);
86 result = mw.commitWriters(cacheFlushSeqNum, false);
87 success = true;
88 }
89 } finally {
90 if (!success && (mw != null)) {
91 for (Path leftoverFile : mw.abortWriters()) {
92 try {
93 store.getFileSystem().delete(leftoverFile, false);
94 } catch (Exception e) {
95 LOG.error("Failed to delete a file after failed flush: " + e);
96 }
97 }
98 }
99 try {
100 scanner.close();
101 } catch (IOException ex) {
102 LOG.warn("Failed to close flush scanner, ignoring", ex);
103 }
104 }
105 return result;
106 }
107
108 private StripeMultiFileWriter.WriterFactory createWriterFactory(
109 final TimeRangeTracker tracker, final long kvCount) {
110 return new StripeMultiFileWriter.WriterFactory() {
111 @Override
112 public Writer createWriter() throws IOException {
113 StoreFile.Writer writer = store.createWriterInTmp(
114 kvCount, store.getFamily().getCompression(),
115
116
117
118
119 tracker);
120 return writer;
121 }
122 };
123 }
124
125
126 public static class StripeFlushRequest {
127
128 protected final KVComparator comparator;
129
130 public StripeFlushRequest(KVComparator comparator) {
131 this.comparator = comparator;
132 }
133
134 @VisibleForTesting
135 public StripeMultiFileWriter createWriter() throws IOException {
136 StripeMultiFileWriter writer = new StripeMultiFileWriter.SizeMultiWriter(comparator, 1,
137 Long.MAX_VALUE, OPEN_KEY, OPEN_KEY);
138 writer.setNoStripeMetadata();
139 return writer;
140 }
141 }
142
143
144 public static class BoundaryStripeFlushRequest extends StripeFlushRequest {
145 private final List<byte[]> targetBoundaries;
146
147
148 public BoundaryStripeFlushRequest(KVComparator comparator, List<byte[]> targetBoundaries) {
149 super(comparator);
150 this.targetBoundaries = targetBoundaries;
151 }
152
153 @Override
154 public StripeMultiFileWriter createWriter() throws IOException {
155 return new StripeMultiFileWriter.BoundaryMultiWriter(comparator, targetBoundaries, null,
156 null);
157 }
158 }
159
160
161 public static class SizeStripeFlushRequest extends StripeFlushRequest {
162 private final int targetCount;
163 private final long targetKvs;
164
165
166
167
168
169
170 public SizeStripeFlushRequest(KVComparator comparator, int targetCount, long targetKvs) {
171 super(comparator);
172 this.targetCount = targetCount;
173 this.targetKvs = targetKvs;
174 }
175
176 @Override
177 public StripeMultiFileWriter createWriter() throws IOException {
178 return new StripeMultiFileWriter.SizeMultiWriter(comparator, this.targetCount, this.targetKvs,
179 OPEN_KEY, OPEN_KEY);
180 }
181 }
182 }