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;
20  
21  
22  import static org.junit.Assert.assertEquals;
23  import static org.junit.Assert.assertNotNull;
24  import static org.junit.Assert.assertTrue;
25  import static org.junit.Assert.fail;
26  
27  import java.io.ByteArrayInputStream;
28  import java.io.ByteArrayOutputStream;
29  import java.io.DataInputStream;
30  import java.io.DataOutputStream;
31  import java.io.IOException;
32  import java.util.List;
33  import java.util.Map;
34  import java.util.NavigableSet;
35  import java.util.Set;
36  
37  import org.apache.hadoop.hbase.client.Get;
38  import org.apache.hadoop.hbase.client.Scan;
39  import org.apache.hadoop.hbase.exceptions.DeserializationException;
40  import org.apache.hadoop.hbase.filter.BinaryComparator;
41  import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
42  import org.apache.hadoop.hbase.filter.Filter;
43  import org.apache.hadoop.hbase.filter.PrefixFilter;
44  import org.apache.hadoop.hbase.filter.RowFilter;
45  import org.apache.hadoop.hbase.io.TimeRange;
46  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
47  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
48  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
49  import org.apache.hadoop.hbase.util.Bytes;
50  import org.apache.hadoop.hbase.util.Writables;
51  import org.apache.hadoop.io.DataInputBuffer;
52  import org.junit.Test;
53  import org.junit.experimental.categories.Category;
54  
55  /**
56   * Test HBase Writables serializations
57   */
58  @Category(SmallTests.class)
59  public class TestSerialization {
60    @Test public void testKeyValue() throws Exception {
61      final String name = "testKeyValue2";
62      byte[] row = name.getBytes();
63      byte[] fam = "fam".getBytes();
64      byte[] qf = "qf".getBytes();
65      long ts = System.currentTimeMillis();
66      byte[] val = "val".getBytes();
67      KeyValue kv = new KeyValue(row, fam, qf, ts, val);
68      ByteArrayOutputStream baos = new ByteArrayOutputStream();
69      DataOutputStream dos = new DataOutputStream(baos);
70      long l = KeyValue.write(kv, dos);
71      dos.close();
72      byte [] mb = baos.toByteArray();
73      ByteArrayInputStream bais = new ByteArrayInputStream(mb);
74      DataInputStream dis = new DataInputStream(bais);
75      KeyValue deserializedKv = KeyValue.create(dis);
76      assertTrue(Bytes.equals(kv.getBuffer(), deserializedKv.getBuffer()));
77      assertEquals(kv.getOffset(), deserializedKv.getOffset());
78      assertEquals(kv.getLength(), deserializedKv.getLength());
79    }
80  
81    @Test public void testCreateKeyValueInvalidNegativeLength() {
82  
83      KeyValue kv_0 = new KeyValue(Bytes.toBytes("myRow"), Bytes.toBytes("myCF"),       // 51 bytes
84                                   Bytes.toBytes("myQualifier"), 12345L, Bytes.toBytes("my12345"));
85  
86      KeyValue kv_1 = new KeyValue(Bytes.toBytes("myRow"), Bytes.toBytes("myCF"),       // 49 bytes
87                                   Bytes.toBytes("myQualifier"), 12345L, Bytes.toBytes("my123"));
88  
89      ByteArrayOutputStream baos = new ByteArrayOutputStream();
90      DataOutputStream dos = new DataOutputStream(baos);
91  
92      long l = 0;
93      try {
94        l  = KeyValue.oswrite(kv_0, dos, false);
95        l += KeyValue.oswrite(kv_1, dos, false);
96        assertEquals(100L, l);
97      } catch (IOException e) {
98        fail("Unexpected IOException" + e.getMessage());
99      }
100 
101     ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
102     DataInputStream dis = new DataInputStream(bais);
103 
104     try {
105       KeyValue.create(dis);
106       assertTrue(kv_0.equals(kv_1));
107     } catch (Exception e) {
108       fail("Unexpected Exception" + e.getMessage());
109     }
110 
111     // length -1
112     try {
113       // even if we have a good kv now in dis we will just pass length with -1 for simplicity
114       KeyValue.create(-1, dis);
115       fail("Expected corrupt stream");
116     } catch (Exception e) {
117       assertEquals("Failed read -1 bytes, stream corrupt?", e.getMessage());
118     }
119 
120   }
121 
122   @Test
123   public void testSplitLogTask() throws DeserializationException {
124     SplitLogTask slt = new SplitLogTask.Unassigned(ServerName.valueOf("mgr,1,1"), 
125       RecoveryMode.LOG_REPLAY);
126     byte [] bytes = slt.toByteArray();
127     SplitLogTask sltDeserialized = SplitLogTask.parseFrom(bytes);
128     assertTrue(slt.equals(sltDeserialized));
129   }
130 
131   @Test public void testCompareFilter() throws Exception {
132     Filter f = new RowFilter(CompareOp.EQUAL,
133       new BinaryComparator(Bytes.toBytes("testRowOne-2")));
134     byte [] bytes = f.toByteArray();
135     Filter ff = RowFilter.parseFrom(bytes);
136     assertNotNull(ff);
137   }
138 
139   @Test public void testTableDescriptor() throws Exception {
140     final String name = "testTableDescriptor";
141     HTableDescriptor htd = createTableDescriptor(name);
142     byte [] mb = Writables.getBytes(htd);
143     HTableDescriptor deserializedHtd =
144       (HTableDescriptor)Writables.getWritable(mb, new HTableDescriptor());
145     assertEquals(htd.getTableName(), deserializedHtd.getTableName());
146   }
147 
148   /**
149    * Test RegionInfo serialization
150    * @throws Exception
151    */
152   @Test public void testRegionInfo() throws Exception {
153     HRegionInfo hri = createRandomRegion("testRegionInfo");
154 
155     //test toByteArray()
156     byte [] hrib = hri.toByteArray();
157     HRegionInfo deserializedHri = HRegionInfo.parseFrom(hrib);
158     assertEquals(hri.getEncodedName(), deserializedHri.getEncodedName());
159     assertEquals(hri, deserializedHri);
160 
161     //test toDelimitedByteArray()
162     hrib = hri.toDelimitedByteArray();
163     DataInputBuffer buf = new DataInputBuffer();
164     try {
165       buf.reset(hrib, hrib.length);
166       deserializedHri = HRegionInfo.parseFrom(buf);
167       assertEquals(hri.getEncodedName(), deserializedHri.getEncodedName());
168       assertEquals(hri, deserializedHri);
169     } finally {
170       buf.close();
171     }
172   }
173 
174   @Test public void testRegionInfos() throws Exception {
175     HRegionInfo hri = createRandomRegion("testRegionInfos");
176     byte[] triple = HRegionInfo.toDelimitedByteArray(hri, hri, hri);
177     List<HRegionInfo> regions = HRegionInfo.parseDelimitedFrom(triple, 0, triple.length);
178     assertTrue(regions.size() == 3);
179     assertTrue(regions.get(0).equals(regions.get(1)));
180     assertTrue(regions.get(0).equals(regions.get(2)));
181   }
182 
183   private HRegionInfo createRandomRegion(final String name) {
184     HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(name));
185     String [] families = new String [] {"info", "anchor"};
186     for (int i = 0; i < families.length; i++) {
187       htd.addFamily(new HColumnDescriptor(families[i]));
188     }
189     return new HRegionInfo(htd.getTableName(), HConstants.EMPTY_START_ROW,
190       HConstants.EMPTY_END_ROW);
191   }
192 
193   /*
194    * TODO
195   @Test public void testPut() throws Exception{
196     byte[] row = "row".getBytes();
197     byte[] fam = "fam".getBytes();
198     byte[] qf1 = "qf1".getBytes();
199     byte[] qf2 = "qf2".getBytes();
200     byte[] qf3 = "qf3".getBytes();
201     byte[] qf4 = "qf4".getBytes();
202     byte[] qf5 = "qf5".getBytes();
203     byte[] qf6 = "qf6".getBytes();
204     byte[] qf7 = "qf7".getBytes();
205     byte[] qf8 = "qf8".getBytes();
206 
207     long ts = System.currentTimeMillis();
208     byte[] val = "val".getBytes();
209 
210     Put put = new Put(row);
211     put.setWriteToWAL(false);
212     put.add(fam, qf1, ts, val);
213     put.add(fam, qf2, ts, val);
214     put.add(fam, qf3, ts, val);
215     put.add(fam, qf4, ts, val);
216     put.add(fam, qf5, ts, val);
217     put.add(fam, qf6, ts, val);
218     put.add(fam, qf7, ts, val);
219     put.add(fam, qf8, ts, val);
220 
221     byte[] sb = Writables.getBytes(put);
222     Put desPut = (Put)Writables.getWritable(sb, new Put());
223 
224     //Timing test
225 //    long start = System.nanoTime();
226 //    desPut = (Put)Writables.getWritable(sb, new Put());
227 //    long stop = System.nanoTime();
228 //    System.out.println("timer " +(stop-start));
229 
230     assertTrue(Bytes.equals(put.getRow(), desPut.getRow()));
231     List<KeyValue> list = null;
232     List<KeyValue> desList = null;
233     for(Map.Entry<byte[], List<KeyValue>> entry : put.getFamilyMap().entrySet()){
234       assertTrue(desPut.getFamilyMap().containsKey(entry.getKey()));
235       list = entry.getValue();
236       desList = desPut.getFamilyMap().get(entry.getKey());
237       for(int i=0; i<list.size(); i++){
238         assertTrue(list.get(i).equals(desList.get(i)));
239       }
240     }
241   }
242 
243 
244   @Test public void testPut2() throws Exception{
245     byte[] row = "testAbort,,1243116656250".getBytes();
246     byte[] fam = "historian".getBytes();
247     byte[] qf1 = "creation".getBytes();
248 
249     long ts = 9223372036854775807L;
250     byte[] val = "dont-care".getBytes();
251 
252     Put put = new Put(row);
253     put.add(fam, qf1, ts, val);
254 
255     byte[] sb = Writables.getBytes(put);
256     Put desPut = (Put)Writables.getWritable(sb, new Put());
257 
258     assertTrue(Bytes.equals(put.getRow(), desPut.getRow()));
259     List<KeyValue> list = null;
260     List<KeyValue> desList = null;
261     for(Map.Entry<byte[], List<KeyValue>> entry : put.getFamilyMap().entrySet()){
262       assertTrue(desPut.getFamilyMap().containsKey(entry.getKey()));
263       list = entry.getValue();
264       desList = desPut.getFamilyMap().get(entry.getKey());
265       for(int i=0; i<list.size(); i++){
266         assertTrue(list.get(i).equals(desList.get(i)));
267       }
268     }
269   }
270 
271 
272   @Test public void testDelete() throws Exception{
273     byte[] row = "row".getBytes();
274     byte[] fam = "fam".getBytes();
275     byte[] qf1 = "qf1".getBytes();
276 
277     long ts = System.currentTimeMillis();
278 
279     Delete delete = new Delete(row);
280     delete.deleteColumn(fam, qf1, ts);
281 
282     byte[] sb = Writables.getBytes(delete);
283     Delete desDelete = (Delete)Writables.getWritable(sb, new Delete());
284 
285     assertTrue(Bytes.equals(delete.getRow(), desDelete.getRow()));
286     List<KeyValue> list = null;
287     List<KeyValue> desList = null;
288     for(Map.Entry<byte[], List<KeyValue>> entry :
289         delete.getFamilyMap().entrySet()){
290       assertTrue(desDelete.getFamilyMap().containsKey(entry.getKey()));
291       list = entry.getValue();
292       desList = desDelete.getFamilyMap().get(entry.getKey());
293       for(int i=0; i<list.size(); i++){
294         assertTrue(list.get(i).equals(desList.get(i)));
295       }
296     }
297   }
298   */
299 
300   @Test public void testGet() throws Exception{
301     byte[] row = "row".getBytes();
302     byte[] fam = "fam".getBytes();
303     byte[] qf1 = "qf1".getBytes();
304 
305     long ts = System.currentTimeMillis();
306     int maxVersions = 2;
307 
308     Get get = new Get(row);
309     get.addColumn(fam, qf1);
310     get.setTimeRange(ts, ts+1);
311     get.setMaxVersions(maxVersions);
312 
313     ClientProtos.Get getProto = ProtobufUtil.toGet(get);
314     Get desGet = ProtobufUtil.toGet(getProto);
315 
316     assertTrue(Bytes.equals(get.getRow(), desGet.getRow()));
317     Set<byte[]> set = null;
318     Set<byte[]> desSet = null;
319 
320     for(Map.Entry<byte[], NavigableSet<byte[]>> entry :
321         get.getFamilyMap().entrySet()){
322       assertTrue(desGet.getFamilyMap().containsKey(entry.getKey()));
323       set = entry.getValue();
324       desSet = desGet.getFamilyMap().get(entry.getKey());
325       for(byte [] qualifier : set){
326         assertTrue(desSet.contains(qualifier));
327       }
328     }
329 
330     assertEquals(get.getMaxVersions(), desGet.getMaxVersions());
331     TimeRange tr = get.getTimeRange();
332     TimeRange desTr = desGet.getTimeRange();
333     assertEquals(tr.getMax(), desTr.getMax());
334     assertEquals(tr.getMin(), desTr.getMin());
335   }
336 
337 
338   @Test public void testScan() throws Exception {
339 
340     byte[] startRow = "startRow".getBytes();
341     byte[] stopRow  = "stopRow".getBytes();
342     byte[] fam = "fam".getBytes();
343     byte[] qf1 = "qf1".getBytes();
344 
345     long ts = System.currentTimeMillis();
346     int maxVersions = 2;
347 
348     Scan scan = new Scan(startRow, stopRow);
349     scan.addColumn(fam, qf1);
350     scan.setTimeRange(ts, ts+1);
351     scan.setMaxVersions(maxVersions);
352 
353     ClientProtos.Scan scanProto = ProtobufUtil.toScan(scan);
354     Scan desScan = ProtobufUtil.toScan(scanProto);
355 
356     assertTrue(Bytes.equals(scan.getStartRow(), desScan.getStartRow()));
357     assertTrue(Bytes.equals(scan.getStopRow(), desScan.getStopRow()));
358     assertEquals(scan.getCacheBlocks(), desScan.getCacheBlocks());
359     Set<byte[]> set = null;
360     Set<byte[]> desSet = null;
361 
362     for(Map.Entry<byte[], NavigableSet<byte[]>> entry :
363         scan.getFamilyMap().entrySet()){
364       assertTrue(desScan.getFamilyMap().containsKey(entry.getKey()));
365       set = entry.getValue();
366       desSet = desScan.getFamilyMap().get(entry.getKey());
367       for(byte[] column : set){
368         assertTrue(desSet.contains(column));
369       }
370 
371       // Test filters are serialized properly.
372       scan = new Scan(startRow);
373       final String name = "testScan";
374       byte [] prefix = Bytes.toBytes(name);
375       scan.setFilter(new PrefixFilter(prefix));
376       scanProto = ProtobufUtil.toScan(scan);
377       desScan = ProtobufUtil.toScan(scanProto);
378       Filter f = desScan.getFilter();
379       assertTrue(f instanceof PrefixFilter);
380     }
381 
382     assertEquals(scan.getMaxVersions(), desScan.getMaxVersions());
383     TimeRange tr = scan.getTimeRange();
384     TimeRange desTr = desScan.getTimeRange();
385     assertEquals(tr.getMax(), desTr.getMax());
386     assertEquals(tr.getMin(), desTr.getMin());
387   }
388 
389   /*
390    * TODO
391   @Test public void testResultEmpty() throws Exception {
392     List<KeyValue> keys = new ArrayList<KeyValue>();
393     Result r = Result.newResult(keys);
394     assertTrue(r.isEmpty());
395     byte [] rb = Writables.getBytes(r);
396     Result deserializedR = (Result)Writables.getWritable(rb, new Result());
397     assertTrue(deserializedR.isEmpty());
398   }
399 
400 
401   @Test public void testResult() throws Exception {
402     byte [] rowA = Bytes.toBytes("rowA");
403     byte [] famA = Bytes.toBytes("famA");
404     byte [] qfA = Bytes.toBytes("qfA");
405     byte [] valueA = Bytes.toBytes("valueA");
406 
407     byte [] rowB = Bytes.toBytes("rowB");
408     byte [] famB = Bytes.toBytes("famB");
409     byte [] qfB = Bytes.toBytes("qfB");
410     byte [] valueB = Bytes.toBytes("valueB");
411 
412     KeyValue kvA = new KeyValue(rowA, famA, qfA, valueA);
413     KeyValue kvB = new KeyValue(rowB, famB, qfB, valueB);
414 
415     Result result = Result.newResult(new KeyValue[]{kvA, kvB});
416 
417     byte [] rb = Writables.getBytes(result);
418     Result deResult = (Result)Writables.getWritable(rb, new Result());
419 
420     assertTrue("results are not equivalent, first key mismatch",
421         result.raw()[0].equals(deResult.raw()[0]));
422 
423     assertTrue("results are not equivalent, second key mismatch",
424         result.raw()[1].equals(deResult.raw()[1]));
425 
426     // Test empty Result
427     Result r = new Result();
428     byte [] b = Writables.getBytes(r);
429     Result deserialized = (Result)Writables.getWritable(b, new Result());
430     assertEquals(r.size(), deserialized.size());
431   }
432 
433   @Test public void testResultDynamicBuild() throws Exception {
434     byte [] rowA = Bytes.toBytes("rowA");
435     byte [] famA = Bytes.toBytes("famA");
436     byte [] qfA = Bytes.toBytes("qfA");
437     byte [] valueA = Bytes.toBytes("valueA");
438 
439     byte [] rowB = Bytes.toBytes("rowB");
440     byte [] famB = Bytes.toBytes("famB");
441     byte [] qfB = Bytes.toBytes("qfB");
442     byte [] valueB = Bytes.toBytes("valueB");
443 
444     KeyValue kvA = new KeyValue(rowA, famA, qfA, valueA);
445     KeyValue kvB = new KeyValue(rowB, famB, qfB, valueB);
446 
447     Result result = Result.newResult(new KeyValue[]{kvA, kvB});
448 
449     byte [] rb = Writables.getBytes(result);
450 
451 
452     // Call getRow() first
453     Result deResult = (Result)Writables.getWritable(rb, new Result());
454     byte [] row = deResult.getRow();
455     assertTrue(Bytes.equals(row, rowA));
456 
457     // Call sorted() first
458     deResult = (Result)Writables.getWritable(rb, new Result());
459     assertTrue("results are not equivalent, first key mismatch",
460         result.raw()[0].equals(deResult.raw()[0]));
461     assertTrue("results are not equivalent, second key mismatch",
462         result.raw()[1].equals(deResult.raw()[1]));
463 
464     // Call raw() first
465     deResult = (Result)Writables.getWritable(rb, new Result());
466     assertTrue("results are not equivalent, first key mismatch",
467         result.raw()[0].equals(deResult.raw()[0]));
468     assertTrue("results are not equivalent, second key mismatch",
469         result.raw()[1].equals(deResult.raw()[1]));
470 
471 
472   }
473 
474   @Test public void testResultArray() throws Exception {
475     byte [] rowA = Bytes.toBytes("rowA");
476     byte [] famA = Bytes.toBytes("famA");
477     byte [] qfA = Bytes.toBytes("qfA");
478     byte [] valueA = Bytes.toBytes("valueA");
479 
480     byte [] rowB = Bytes.toBytes("rowB");
481     byte [] famB = Bytes.toBytes("famB");
482     byte [] qfB = Bytes.toBytes("qfB");
483     byte [] valueB = Bytes.toBytes("valueB");
484 
485     KeyValue kvA = new KeyValue(rowA, famA, qfA, valueA);
486     KeyValue kvB = new KeyValue(rowB, famB, qfB, valueB);
487 
488 
489     Result result1 = Result.newResult(new KeyValue[]{kvA, kvB});
490     Result result2 = Result.newResult(new KeyValue[]{kvB});
491     Result result3 = Result.newResult(new KeyValue[]{kvB});
492 
493     Result [] results = new Result [] {result1, result2, result3};
494 
495     ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
496     DataOutputStream out = new DataOutputStream(byteStream);
497     Result.writeArray(out, results);
498 
499     byte [] rb = byteStream.toByteArray();
500 
501     DataInputBuffer in = new DataInputBuffer();
502     in.reset(rb, 0, rb.length);
503 
504     Result [] deResults = Result.readArray(in);
505 
506     assertTrue(results.length == deResults.length);
507 
508     for(int i=0;i<results.length;i++) {
509       KeyValue [] keysA = results[i].raw();
510       KeyValue [] keysB = deResults[i].raw();
511       assertTrue(keysA.length == keysB.length);
512       for(int j=0;j<keysA.length;j++) {
513         assertTrue("Expected equivalent keys but found:\n" +
514             "KeyA : " + keysA[j].toString() + "\n" +
515             "KeyB : " + keysB[j].toString() + "\n" +
516             keysA.length + " total keys, " + i + "th so far"
517             ,keysA[j].equals(keysB[j]));
518       }
519     }
520 
521   }
522 
523   @Test public void testResultArrayEmpty() throws Exception {
524     List<KeyValue> keys = new ArrayList<KeyValue>();
525     Result r = Result.newResult(keys);
526     Result [] results = new Result [] {r};
527 
528     ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
529     DataOutputStream out = new DataOutputStream(byteStream);
530 
531     Result.writeArray(out, results);
532 
533     results = null;
534 
535     byteStream = new ByteArrayOutputStream();
536     out = new DataOutputStream(byteStream);
537     Result.writeArray(out, results);
538 
539     byte [] rb = byteStream.toByteArray();
540 
541     DataInputBuffer in = new DataInputBuffer();
542     in.reset(rb, 0, rb.length);
543 
544     Result [] deResults = Result.readArray(in);
545 
546     assertTrue(deResults.length == 0);
547 
548     results = new Result[0];
549 
550     byteStream = new ByteArrayOutputStream();
551     out = new DataOutputStream(byteStream);
552     Result.writeArray(out, results);
553 
554     rb = byteStream.toByteArray();
555 
556     in = new DataInputBuffer();
557     in.reset(rb, 0, rb.length);
558 
559     deResults = Result.readArray(in);
560 
561     assertTrue(deResults.length == 0);
562 
563   }
564   */
565 
566   protected static final int MAXVERSIONS = 3;
567   protected final static byte [] fam1 = Bytes.toBytes("colfamily1");
568   protected final static byte [] fam2 = Bytes.toBytes("colfamily2");
569   protected final static byte [] fam3 = Bytes.toBytes("colfamily3");
570   protected static final byte [][] COLUMNS = {fam1, fam2, fam3};
571 
572   /**
573    * Create a table of name <code>name</code> with {@link COLUMNS} for
574    * families.
575    * @param name Name to give table.
576    * @return Column descriptor.
577    */
578   protected HTableDescriptor createTableDescriptor(final String name) {
579     return createTableDescriptor(name, MAXVERSIONS);
580   }
581 
582   /**
583    * Create a table of name <code>name</code> with {@link COLUMNS} for
584    * families.
585    * @param name Name to give table.
586    * @param versions How many versions to allow per column.
587    * @return Column descriptor.
588    */
589   protected HTableDescriptor createTableDescriptor(final String name,
590       final int versions) {
591     HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(name));
592     htd.addFamily(new HColumnDescriptor(fam1)
593         .setMaxVersions(versions)
594         .setBlockCacheEnabled(false)
595     );
596     htd.addFamily(new HColumnDescriptor(fam2)
597         .setMaxVersions(versions)
598         .setBlockCacheEnabled(false)
599     );
600     htd.addFamily(new HColumnDescriptor(fam3)
601         .setMaxVersions(versions)
602         .setBlockCacheEnabled(false)
603     );
604     return htd;
605   }
606 }