in
+ * @param in Where to read bytes from. Creates a byte array to hold the KeyValue
+ * backing bytes copied from the steam.
+ * @return KeyValue created by deserializing from in OR if we find a length
+ * of zero, we will return null which can be useful marking a stream as done.
* @throws IOException
*/
public static KeyValue create(final DataInput in) throws IOException {
@@ -2301,7 +2303,8 @@ public class KeyValue implements Cell, HeapSize {
* Create a KeyValue reading length from in
* @param length
* @param in
- * @return Created KeyValue
+ * @return Created KeyValue OR if we find a length of zero, we will return null which
+ * can be useful marking a stream as done.
* @throws IOException
*/
public static KeyValue create(int length, final DataInput in) throws IOException {
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
index a2f75b9..543b5e4 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
@@ -28,7 +28,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.IterableUtils;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hbase.Cell;
-import org.apache.hbase.cell.CellTool;
+import org.apache.hbase.CellTool;
/**
* static convenience methods for dealing with KeyValues and collections of KeyValues
diff --git a/hbase-common/src/main/java/org/apache/hbase/Cell.java b/hbase-common/src/main/java/org/apache/hbase/Cell.java
index 8247bab..3a57e8b 100644
--- a/hbase-common/src/main/java/org/apache/hbase/Cell.java
+++ b/hbase-common/src/main/java/org/apache/hbase/Cell.java
@@ -20,7 +20,6 @@ package org.apache.hbase;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hbase.cell.CellTool;
/**
diff --git a/hbase-common/src/main/java/org/apache/hbase/CellComparator.java b/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
new file mode 100644
index 0000000..f6a3c25
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hbase;
+
+import java.io.Serializable;
+import java.util.Comparator;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import com.google.common.primitives.Longs;
+
+/**
+ * Compare two traditional HBase cells.
+ *
+ * Note: This comparator is not valid for -ROOT- and .META. tables.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class CellComparator implements Comparator+ * The {@link #get()} method will return a reference to a Cell implementation. This reference may + * or may not point to a reusable cell implementation, so users of the CellScanner should not, for + * example, accumulate a List of Cells. All of the references may point to the same object, which + * would be the latest state of the underlying Cell. In short, the Cell is mutable. Use + * {@link #getDeepCopy()} if you want a Cell that does not hold references (if the data is + * encoded or compressed, the call to {@link #getDeepCopy()} will cost more than a {@link #get()}. + *
+ * Typical usage: + * + *
+ * while (scanner.next()) {
+ * Cell cell = scanner.get();
+ * // do something
+ * }
+ *
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface CellScanner {
+ /**
+ * @return the current Cell which may be mutable
+ */
+ Cell get();
+
+ /**
+ * @return Get a deep copy of the current Cell or null if no current Cell
+ */
+ Cell getDeepCopy();
+
+ /**
+ * Advance the scanner 1 cell.
+ * @return true if the next cell is found and {@link #get()} will return a valid Cell
+ */
+ boolean next();
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/CellTool.java b/hbase-common/src/main/java/org/apache/hbase/CellTool.java
new file mode 100644
index 0000000..96bb509
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/CellTool.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hbase;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.util.ByteRange;
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class CellTool {
+
+ /******************* ByteRange *******************************/
+
+ public static ByteRange fillRowRange(Cell cell, ByteRange range) {
+ return range.set(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
+ }
+
+ public static ByteRange fillFamilyRange(Cell cell, ByteRange range) {
+ return range.set(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
+ }
+
+ public static ByteRange fillQualifierRange(Cell cell, ByteRange range) {
+ return range.set(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength());
+ }
+
+
+ /***************** get individual arrays for tests ************/
+
+ public static byte[] getRowArray(Cell cell){
+ byte[] output = new byte[cell.getRowLength()];
+ copyRowTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getFamilyArray(Cell cell){
+ byte[] output = new byte[cell.getFamilyLength()];
+ copyFamilyTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getQualifierArray(Cell cell){
+ byte[] output = new byte[cell.getQualifierLength()];
+ copyQualifierTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getValueArray(Cell cell){
+ byte[] output = new byte[cell.getValueLength()];
+ copyValueTo(cell, output, 0);
+ return output;
+ }
+
+
+ /******************** copyTo **********************************/
+
+ public static int copyRowTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getRowArray(), cell.getRowOffset(), destination, destinationOffset,
+ cell.getRowLength());
+ return destinationOffset + cell.getRowLength();
+ }
+
+ public static int copyFamilyTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getFamilyArray(), cell.getFamilyOffset(), destination, destinationOffset,
+ cell.getFamilyLength());
+ return destinationOffset + cell.getFamilyLength();
+ }
+
+ public static int copyQualifierTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getQualifierArray(), cell.getQualifierOffset(), destination,
+ destinationOffset, cell.getQualifierLength());
+ return destinationOffset + cell.getQualifierLength();
+ }
+
+ public static int copyValueTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getValueArray(), cell.getValueOffset(), destination, destinationOffset,
+ cell.getValueLength());
+ return destinationOffset + cell.getValueLength();
+ }
+
+
+ /********************* misc *************************************/
+
+ public static byte getRowByte(Cell cell, int index) {
+ return cell.getRowArray()[cell.getRowOffset() + index];
+ }
+
+
+ /********************** KeyValue (move to KeyValueUtils) *********************/
+
+ public static ByteBuffer getValueBufferShallowCopy(Cell cell) {
+ ByteBuffer buffer = ByteBuffer.wrap(cell.getValueArray(), cell.getValueOffset(),
+ cell.getValueLength());
+// buffer.position(buffer.limit());//make it look as if value was appended
+ return buffer;
+ }
+
+}
diff --git a/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java b/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java
deleted file mode 100644
index 19d6b54..0000000
--- a/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hbase.cell;
-
-import java.io.Serializable;
-import java.util.Comparator;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hbase.Cell;
-
-import com.google.common.primitives.Longs;
-
-/**
- * Compare two traditional HBase cells.
- *
- * Note: This comparator is not valid for -ROOT- and .META. tables.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class CellComparator implements ComparatorIf you wrote two KeyValues to this encoder, it would look like this in the stream: + *
+ * length-of-KeyValue1 // A java int with the length of KeyValue1 backing array + * KeyValue1 backing array filled with a KeyValue serialized in its particular format + * length-of-KeyValue2 + * KeyValue2 backing array + * length-of-zero // A java int whose value is 0; marks end of the encoded section + *+ * @see OldSchoolKeyValueDecoder + */ +public class OldSchoolKeyValueEncoder implements CellOutputStream { + // We write out an '0' int as marker that there are no more kvs when you call flush. + static final int END_OF_KEYVALUES = 0; + // Need to be able to write java types such as int and long so need DataOutput. + // Want to stream too so DataOutputStream. + private final DataOutputStream out; + // This encoder is 'done' once flush has been called because on flush we + // write out the END_OF_KEYVALUES marker. + private boolean flush = false; + + public OldSchoolKeyValueEncoder(final DataOutputStream out) { + this.out = out; + } + + @Override + public void write(Cell cell) throws IOException { + if (this.flush) throw new CodecException("Finished"); + // This is crass and will not work when KV changes + try { + KeyValue.write((KeyValue)cell, this.out); + } catch (IOException e) { + throw new CodecException(e); + } + } + + /** + * Calling flush 'finishes' this encoder. Subsequent calls + * to {@link #write(Cell)} will throw exception. + */ + @Override + public void flush() throws IOException { + if (this.flush) return; + this.flush = true; + // Write out an int whose value is zero as end of stream. + try { + this.out.writeInt(END_OF_KEYVALUES); + this.out.flush(); + } catch (IOException e) { + throw new CodecException(e); + } + } +} \ No newline at end of file diff --git a/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java b/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java new file mode 100644 index 0000000..b42005f --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hbase.io; + +import java.io.IOException; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hbase.Cell; + +/** + * Accepts a stream of Cells. This can be used to build a block of cells during compactions + * and flushes, or to build a byte[] to send to the client. This could be backed by a + * List
To read Cells, use {@link CellInputStream} + * @see CellInputStream + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +public interface CellOutputStream { + /** + * Implementation must copy the entire state of the Cell. If the written Cell is modified + * immediately after the write method returns, the modifications must have absolutely no effect + * on the copy of the Cell that was added in the write. + * @param cell Cell to write out + * @throws IOException + */ + void write(Cell cell) throws IOException; + + /** + * Let the implementation decide what to do. Usually means writing accumulated data into a byte[] + * that can then be read from the implementation to be sent to disk, put in the block cache, or + * sent over the network. + * @throws IOException + */ + void flush() throws IOException; +} \ No newline at end of file diff --git a/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java b/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java new file mode 100644 index 0000000..458bd8a --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java @@ -0,0 +1,93 @@ +package org.apache.hbase.codec; + +import static org.junit.Assert.*; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hbase.Cell; +import org.apache.hbase.CellComparator; +import org.apache.hbase.codec.CellDecoder; +import org.apache.hbase.codec.CellEncoder; +import org.junit.Test; + +import com.google.common.io.CountingInputStream; +import com.google.common.io.CountingOutputStream; + +public class TestCellCodec { + + @Test + public void testEmptyWorks() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + CellEncoder encoder = new CellEncoder(dos); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + assertEquals(0 + 1 /*CellEncoder.END_OF_CELLS*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + assertFalse(decoder.next()); + dis.close(); + assertEquals(0 + 1 /*CellEncoder.END_OF_CELLS*/, cis.getCount()); + } + + @Test + public void testOne() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + CellEncoder encoder = new CellEncoder(dos); + final KeyValue kv = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v")); + encoder.write(kv); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + assertTrue(decoder.next()); // First read should pull in the KV + assertFalse(decoder.next()); // Second read should trip over the end-of-stream marker and return false + dis.close(); + assertEquals(offset, cis.getCount()); + } + + @Test + public void testThree() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + CellEncoder encoder = new CellEncoder(dos); + final KeyValue kv1 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("1"), Bytes.toBytes("1")); + final KeyValue kv2 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("2"), Bytes.toBytes("2")); + final KeyValue kv3 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("3"), Bytes.toBytes("3")); + encoder.write(kv1); + encoder.write(kv2); + encoder.write(kv3); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + assertTrue(decoder.next()); + Cell c = decoder.get(); + assertTrue(CellComparator.equals(c, kv1)); + assertTrue(decoder.next()); + c = decoder.get(); + assertTrue(CellComparator.equals(c, kv2)); + assertTrue(decoder.next()); + c = decoder.get(); + assertTrue(CellComparator.equals(c, kv3)); + assertFalse(decoder.next()); + dis.close(); + assertEquals(offset, cis.getCount()); + } +} \ No newline at end of file diff --git a/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java b/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java new file mode 100644 index 0000000..af788c1 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java @@ -0,0 +1,98 @@ +package org.apache.hbase.codec; + +import static org.junit.Assert.*; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hbase.codec.OldSchoolKeyValueDecoder; +import org.apache.hbase.codec.OldSchoolKeyValueEncoder; +import org.junit.Test; + +import com.google.common.io.CountingInputStream; +import com.google.common.io.CountingOutputStream; + +public class TestOldSchoolKeyValueCodec { + + @Test + public void testEmptyWorks() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(dos); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + assertEquals(0 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + assertFalse(decoder.next()); + dis.close(); + assertEquals(0 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } + + @Test + public void testOne() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(dos); + final KeyValue kv = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v")); + final long length = kv.getLength() + Bytes.SIZEOF_INT; + encoder.write(kv); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + assertEquals(length + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + assertTrue(decoder.next()); // First read should pull in the KV + assertFalse(decoder.next()); // Second read should trip over the end-of-stream marker and return false + dis.close(); + assertEquals(length + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } + + @Test + public void testThree() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(dos); + final KeyValue kv1 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("1"), Bytes.toBytes("1")); + final KeyValue kv2 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("2"), Bytes.toBytes("2")); + final KeyValue kv3 = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("3"), Bytes.toBytes("3")); + final long length = kv1.getLength() + Bytes.SIZEOF_INT; + encoder.write(kv1); + encoder.write(kv2); + encoder.write(kv3); + encoder.flush(); + dos.close(); + long offset = cos.getCount(); + assertEquals(length * 3 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + assertTrue(decoder.next()); + KeyValue kv = (KeyValue)decoder.get(); + assertTrue(kv1.equals(kv)); + assertTrue(decoder.next()); + kv = (KeyValue)decoder.get(); + assertTrue(kv2.equals(kv)); + assertTrue(decoder.next()); + kv = (KeyValue)decoder.get(); + assertTrue(kv3.equals(kv)); + assertFalse(decoder.next()); + dis.close(); + assertEquals((length * 3) + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } +} \ No newline at end of file