Merge lp:~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment into lp:akiban-persistit

[Peter Beaman]

Add a test to exercise various places where code in the Key#append methods attempts to write past the end of the encoded byte array.

Add try/catch blocks to all relevant Key#append methods to catch AIOOBE and instead throw newly added KeyTooLongException. KeyTooLongException is unchecked so as not to change lots of use cases where Keys modifications occur.

Strategy is to catch the AIOOBE, which is a rare, unexpected event rather than to add bound-checking code everywhere. Since the array operations already do bounds checking this seems like a better approach.

[Nathan Williams (nwilliams)]

We should always restore _size in the even of an error, right? There are a handful of places that don't do that, such as append(boolean). Perhaps adding the size to restore to the new tooLong() helper method would enforce that nicely.

Otherwise seems simple enough.

review: Needs Information

[Peter Beaman (pbeaman)]

I took some care to verify that the places that don't restore the value of
_size also don't change it.

On Wed, Mar 20, 2013 at 2:03 PM, Nathan Williams <email address hidden>wrote:

> Review: Needs Information
>
> We should always restore _size in the even of an error, right? There are a
> handful of places that don't do that, such as append(boolean). Perhaps
> adding the size to restore to the new tooLong() helper method would enforce
> that nicely.
>
> Otherwise seems simple enough.
> --
>
> https://code.launchpad.net/~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment/+merge/154422
> You are the owner of
> lp:~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment.
>

[Peter Beaman (pbeaman)]

Persuaded. Modified all Key#append methods to save starting size and restore it in the tooLong method. As you mentioned, this pattern will help guard against newly added errors.

[Nathan Williams (nwilliams)]

Thanks Peter!

review: Approve

[Peter Beaman (pbeaman)]

Found some new problems and pushed back to Needs Review. Basically, the new KeyTest1#testKeyTooLong does not loop correctly and does not actually test all object values in the TEST_OBJECTS array. In particular, the method Key#appendString did not handle AIOOBE, and that's the one we are more concerned about.

[Akiban Build User (build-akiban)]

There were 2 failures during build/test:

* job server-build failed at build number 3912: http://172.16.20.104:8080/job/server-build/3912/

* view must-pass failed: server-build is aborted

review: Needs Fixing

[lp:~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment]

427. By Peter Beaman on 2013-03-20

Fix testKeyTooLong. Modify appendString to throw KeyTooLongException rather than ConversionException.

[Peter Beaman (pbeaman)]

New modifications:

correct KeyTest1#testKeyTooLongException

add try/catch to appendString, appendNull, appendBefore and appendAfter

[Nathan Williams (nwilliams)]

Looks good, again.

review: Approve

[Peter Beaman (pbeaman)]

Note on a concern I researched a bit. We mangle the keys in the sort tree. If the server were to use those keys to deliver results from a sort or SELECT DISTINCT (like a covering index) then of course we would return the wrong result. However, I believe this does not happen.

It appears the Sort_Tree operator stores and reads data back from value instances stored in the tree, rather than trying to use the key as a source. Specifically, in PValueSorterAdapter$InternalPAdapter, there is a method putToHolders that copies data from a valueSource to the result row. The valueSource is a PersistitValuePValueSource, not a PersistitKeyPValueSource.

The code path is complex, so I would appreciate confirmation that this is the case. But from my reading I can only see a path that reads from a PersistitValuePValueSource.

[Mike McMahon (mmcm)]

> The code path is complex, so I would appreciate confirmation that this is the case. But from my reading I can only see a path that reads from a PersistitValuePValueSource.

Yes, we are good.

Look at Sorter.loadTree() and its calls to createKey() and createValue(). The entire incoming Row is encoded in the value and that is always the output source. This allows it to support cases where you sort on a subset of the columns that pass through. No particular attempt is made to suppress duplication (as you say, like a covering index).

[Peter Beaman (pbeaman)]

The change I made this morning in loadTree (to close the cursor when
there's an Exception) is wrong. Will correct it shortly. I am working on
additional tests.

On Thu, Mar 21, 2013 at 1:24 PM, Mike McMahon <email address hidden> wrote:

> > The code path is complex, so I would appreciate confirmation that this
> is the case. But from my reading I can only see a path that reads from a
> PersistitValuePValueSource.
>
> Yes, we are good.
>
> Look at Sorter.loadTree() and its calls to createKey() and createValue().
> The entire incoming Row is encoded in the value and that is always the
> output source. This allows it to support cases where you sort on a subset
> of the columns that pass through. No particular attempt is made to suppress
> duplication (as you say, like a covering index).
> --
>
> https://code.launchpad.net/~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment/+merge/154422
> You are the owner of
> lp:~pbeaman/akiban-persistit/fix-1157809-aioobe-on-append-key-segment.
>

=== modified file 'src/main/java/com/persistit/Key.java'
--- src/main/java/com/persistit/Key.java	2013-01-19 17:42:22 +0000
+++ src/main/java/com/persistit/Key.java	2013-03-20 19:33:22 +0000
@@ -27,6 +27,7 @@
 import com.persistit.encoding.KeyRenderer;
 import com.persistit.exception.ConversionException;
 import com.persistit.exception.InvalidKeyException;
+import com.persistit.exception.KeyTooLongException;
 import com.persistit.exception.MissingKeySegmentException;
 import com.persistit.util.Util;
 
@@ -924,8 +925,6 @@
      *     Key copiedKey = new Key(originalKey);
      * </pre></code>
      * 
-     * 
-     * 
      * @param key
      *            The <code>Key</code> to copy.
      */
@@ -947,17 +946,25 @@
         key.bumpGeneration();
     }
 
+    /**
+     * Construct a <code>Key</code> with a maximum length of
+     * {@value #MAX_KEY_LENGTH}.
+     * 
+     * @param persistit
+     */
     public Key(final Persistit persistit) {
         this(persistit, MAX_KEY_LENGTH);
     }
 
     /**
-     * Constructs a <code>Key</code> with the specified maximum length.
+     * Construct a <code>Key</code> with the specified maximum length.
      * 
+     * @param Persistit
+     *            the Persistit instance
      * @param maxLength
      *            The maximum length
      */
-    private Key(final Persistit persistit, final int maxLength) {
+    public Key(final Persistit persistit, final int maxLength) {
         _persistit = persistit;
         if (maxLength <= 0) {
             throw new IllegalArgumentException("Key length must be positive");
@@ -1479,13 +1486,11 @@
             } else if (z0 != 0 && z1 == (byte) 1) {
                 nudged = Nudged.RIGHT;
                 _bytes[_size - 1] = 0;
-            } else if (z0 != 0 && z1 != 0) {
+            } else if (z0 != 0 && z1 != 0 && _size < _maxSize) {
                 nudged = Nudged.LEFT;
                 save = _bytes[_size];
                 _bytes[_size] = (byte) 0;
                 _size++;
-                // _bytes[_size - 2]++;
-                // _bytes[_size - 1] = (byte) 0;
             }
         }
 
@@ -1539,11 +1544,15 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final boolean v) {
-        testValidForAppend();
-        int size = _size;
-        _bytes[size++] = v ? (byte) TYPE_BOOLEAN_TRUE : (byte) TYPE_BOOLEAN_FALSE;
-
-        return endSegment(size);
+        final int save = _size;
+        try {
+            testValidForAppend();
+            int size = _size;
+            _bytes[size++] = v ? (byte) TYPE_BOOLEAN_TRUE : (byte) TYPE_BOOLEAN_FALSE;
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /*
@@ -1629,21 +1638,26 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final byte v) {
-        testValidForAppend();
-        int size = _size;
-        if (v > 0) {
-            _bytes[size++] = TYPE_BYTE + EWIDTH_BYTE + 1;
-            _bytes[size++] = (byte) (0x80 | v);
-        } else if (v < 0) {
-            _bytes[size++] = TYPE_BYTE;
-            _bytes[size++] = (byte) (0x80 | v);
-        } else // v == 0
-        {
-            _bytes[size++] = TYPE_BYTE + EWIDTH_BYTE;
+        final int save = _size;
+        try {
+            testValidForAppend();
+            int size = _size;
+            if (v > 0) {
+                _bytes[size++] = TYPE_BYTE + EWIDTH_BYTE + 1;
+                _bytes[size++] = (byte) (0x80 | v);
+            } else if (v < 0) {
+                _bytes[size++] = TYPE_BYTE;
+                _bytes[size++] = (byte) (0x80 | v);
+            } else // v == 0
+            {
+                _bytes[size++] = TYPE_BYTE + EWIDTH_BYTE;
+            }
+            // Close out the segment.
+
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
         }
-        // Close out the segment.
-
-        return endSegment(size);
     }
 
     /**
@@ -1654,9 +1668,65 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final short v) {
-        testValidForAppend();
-        int size = _size;
-        if (v >= 0) {
+        final int save = _size;
+        try {
+            testValidForAppend();
+            int size = _size;
+            if (v >= 0) {
+                int scale = 3;
+                if (v > 0x3FFF)
+                    scale = 0;
+                else if (v > 0x007F)
+                    scale = 1;
+                else if (v > 0x0000)
+                    scale = 2;
+                _bytes[size++] = (byte) (TYPE_SHORT + EWIDTH_SHORT * 2 - scale);
+                switch (scale) {
+                // control falls through intentionally
+                case 0:
+                    _bytes[size++] = (byte) (0x80 | (v >>> 14));
+                case 1:
+                    _bytes[size++] = (byte) (0x80 | (v >>> 7));
+                case 2:
+                    _bytes[size++] = (byte) (0x80 | v);
+                }
+            } else {
+                int scale = 2;
+                if (v < -0x3FFF)
+                    scale = 0;
+                else if (v < -0x007F)
+                    scale = 1;
+                _bytes[size++] = (byte) (TYPE_SHORT + scale);
+                switch (scale) {
+                // control falls through intentionally
+                case 0:
+                    _bytes[size++] = (byte) (0x80 | (v >>> 14));
+                case 1:
+                    _bytes[size++] = (byte) (0x80 | (v >>> 7));
+                case 2:
+                    _bytes[size++] = (byte) (0x80 | v);
+                }
+            }
+            // Close out the segment.
+
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
+    }
+
+    /**
+     * Encodes and appends a char value to the key.
+     * 
+     * @param v
+     *            The char value to append
+     * @return This <code>Key</code>, to permit method call chaining
+     */
+    public Key append(final char v) {
+        final int save = _size;
+        try {
+            testValidForAppend();
+            int size = _size;
             int scale = 3;
             if (v > 0x3FFF)
                 scale = 0;
@@ -1664,67 +1734,21 @@
                 scale = 1;
             else if (v > 0x0000)
                 scale = 2;
-            _bytes[size++] = (byte) (TYPE_SHORT + EWIDTH_SHORT * 2 - scale);
-            switch (scale) {
-            // control falls through intentionally
-            case 0:
-                _bytes[size++] = (byte) (0x80 | (v >>> 14));
-            case 1:
-                _bytes[size++] = (byte) (0x80 | (v >>> 7));
-            case 2:
-                _bytes[size++] = (byte) (0x80 | v);
-            }
-        } else {
-            int scale = 2;
-            if (v < -0x3FFF)
-                scale = 0;
-            else if (v < -0x007F)
-                scale = 1;
-            _bytes[size++] = (byte) (TYPE_SHORT + scale);
-            switch (scale) {
-            // control falls through intentionally
-            case 0:
-                _bytes[size++] = (byte) (0x80 | (v >>> 14));
-            case 1:
-                _bytes[size++] = (byte) (0x80 | (v >>> 7));
-            case 2:
-                _bytes[size++] = (byte) (0x80 | v);
-            }
-        }
-        // Close out the segment.
-
-        return endSegment(size);
-    }
-
-    /**
-     * Encodes and appends a char value to the key.
-     * 
-     * @param v
-     *            The char value to append
-     * @return This <code>Key</code>, to permit method call chaining
-     */
-    public Key append(final char v) {
-        testValidForAppend();
-        int size = _size;
-        int scale = 3;
-        if (v > 0x3FFF)
-            scale = 0;
-        else if (v > 0x007F)
-            scale = 1;
-        else if (v > 0x0000)
-            scale = 2;
-        _bytes[size++] = (byte) (TYPE_CHAR + EWIDTH_CHAR - scale);
-        switch (scale) {
-        // control falls through intentionally
-        case 0:
-            _bytes[size++] = (byte) (0x80 | (v >>> 14));
-        case 1:
-            _bytes[size++] = (byte) (0x80 | (v >>> 7));
-        case 2:
-            _bytes[size++] = (byte) (0x80 | v);
-        }
-
-        return endSegment(size);
+            _bytes[size++] = (byte) (TYPE_CHAR + EWIDTH_CHAR - scale);
+            switch (scale) {
+            // control falls through intentionally
+            case 0:
+                _bytes[size++] = (byte) (0x80 | (v >>> 14));
+            case 1:
+                _bytes[size++] = (byte) (0x80 | (v >>> 7));
+            case 2:
+                _bytes[size++] = (byte) (0x80 | v);
+            }
+
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -1735,11 +1759,14 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final int v) {
-        testValidForAppend();
-        final int size = appendIntInternal(v);
-        // Close out the segment.
-
-        return endSegment(size);
+        final int save = _size;
+        try {
+            testValidForAppend();
+            final int size = appendIntInternal(v);
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -1821,9 +1848,14 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final long v) {
-        testValidForAppend();
-        final int size = appendLongInternal(v);
-        return endSegment(size);
+        final int save = _size;
+        try {
+            testValidForAppend();
+            final int size = appendLongInternal(v);
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private int appendLongInternal(final long v) {
@@ -1921,21 +1953,26 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final float v) {
-        testValidForAppend();
-        int bits = Float.floatToIntBits(v);
-        int size = _size;
-        _bytes[size++] = TYPE_FLOAT;
-        if (bits < 0) {
-            bits = ~bits;
-        } else {
-            bits ^= 0x80000000;
-        }
-        while (bits != 0) {
-            _bytes[size++] = (byte) (0x80 | (bits >> 25));
-            bits <<= 7;
-        }
+        final int save = _size;
+        try {
+            testValidForAppend();
+            int bits = Float.floatToIntBits(v);
+            int size = _size;
+            _bytes[size++] = TYPE_FLOAT;
+            if (bits < 0) {
+                bits = ~bits;
+            } else {
+                bits ^= 0x80000000;
+            }
+            while (bits != 0) {
+                _bytes[size++] = (byte) (0x80 | (bits >> 25));
+                bits <<= 7;
+            }
 
-        return endSegment(size);
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -1946,21 +1983,26 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key append(final double v) {
-        testValidForAppend();
-        long bits = Double.doubleToLongBits(v);
-        int size = _size;
-        _bytes[size++] = TYPE_DOUBLE;
-        if (bits < 0) {
-            bits = ~bits;
-        } else {
-            bits ^= 0x8000000000000000L;
-        }
-        while (bits != 0) {
-            _bytes[size++] = (byte) (0x80 | (bits >> 57));
-            bits <<= 7;
-        }
+        final int save = _size;
+        try {
+            testValidForAppend();
+            long bits = Double.doubleToLongBits(v);
+            int size = _size;
+            _bytes[size++] = TYPE_DOUBLE;
+            if (bits < 0) {
+                bits = ~bits;
+            } else {
+                bits ^= 0x8000000000000000L;
+            }
+            while (bits != 0) {
+                _bytes[size++] = (byte) (0x80 | (bits >> 57));
+                bits <<= 7;
+            }
 
-        return endSegment(size);
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -2072,6 +2114,7 @@
         }
 
         throw new ConversionException("Object class " + object.getClass().getName() + " can't be used in a Key");
+
     }
 
     /**
@@ -2081,16 +2124,22 @@
      * 
      * @param key
      */
-    public void appendKeySegment(final Key key) {
-        int length = 0;
-        for (int index = key.getIndex(); index < key.getEncodedSize(); index++) {
-            length++;
-            if (key.getEncodedBytes()[index] == 0) {
-                break;
+    public Key appendKeySegment(final Key key) {
+        final int save = _size;
+        try {
+            int length = 0;
+            for (int index = key.getIndex(); index < key.getEncodedSize(); index++) {
+                length++;
+                if (key.getEncodedBytes()[index] == 0) {
+                    length--;
+                    break;
+                }
             }
+            System.arraycopy(key.getEncodedBytes(), key.getIndex(), _bytes, _size, length);
+            return endSegment(_size + length);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
         }
-        System.arraycopy(key.getEncodedBytes(), key.getIndex(), _bytes, _size, length);
-        _size += length;
     }
 
     /**
@@ -3325,33 +3374,34 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     private Key appendString(final CharSequence s, final CoderContext context) {
-        notLeftOrRightGuard();
-        testValidForAppend();
-        final int strlen = s.length();
-        if (strlen > _maxSize) {
-            throw new ConversionException("Requested size=" + strlen + " exceeds maximum size=" + _maxSize);
-        }
-        int size = _size;
-        _bytes[size++] = (byte) TYPE_STRING;
+        final int save = _size;
+        try {
+            notLeftOrRightGuard();
+            testValidForAppend();
+            final int strlen = s.length();
+            int size = _size;
+            _bytes[size++] = (byte) TYPE_STRING;
 
-        for (int i = 0; i < strlen; i++) {
-            final int c = s.charAt(i);
-            if (c <= 0x0001) {
-                _bytes[size++] = (byte) (0x01);
-                _bytes[size++] = (byte) (c + 0x0020);
-            } else if (c <= 0x007F) {
-                _bytes[size++] = (byte) c;
-            } else if (c <= 0x07FF) {
-                _bytes[size++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
-                _bytes[size++] = (byte) (0x80 | ((c >> 0) & 0x3F));
-            } else {
-                _bytes[size++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
-                _bytes[size++] = (byte) (0x80 | ((c >> 6) & 0x3F));
-                _bytes[size++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+            for (int i = 0; i < strlen; i++) {
+                final int c = s.charAt(i);
+                if (c <= 0x0001) {
+                    _bytes[size++] = (byte) (0x01);
+                    _bytes[size++] = (byte) (c + 0x0020);
+                } else if (c <= 0x007F) {
+                    _bytes[size++] = (byte) c;
+                } else if (c <= 0x07FF) {
+                    _bytes[size++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
+                    _bytes[size++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+                } else {
+                    _bytes[size++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
+                    _bytes[size++] = (byte) (0x80 | ((c >> 6) & 0x3F));
+                    _bytes[size++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+                }
             }
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
         }
-
-        return endSegment(size);
     }
 
     /**
@@ -3360,9 +3410,14 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     private Key appendNull() {
-        int size = _size;
-        _bytes[size++] = TYPE_NULL;
-        return endSegment(size);
+        final int save = _size;
+        try {
+            int size = _size;
+            _bytes[size++] = TYPE_NULL;
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private Key appendByKeyCoder(final Object object, final Class<?> cl, final KeyCoder coder,
@@ -3379,6 +3434,8 @@
             endSegment(_size);
             size = _size;
             return this;
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(size);
         } finally {
             _size = size;
             _inKeyCoder = saveInKeyCoder;
@@ -3480,14 +3537,19 @@
      * level. The result key value should only be used in traversal operations.
      */
     Key appendBefore() {
-        int size = _size;
-        _bytes[size++] = TYPE_BEFORE;
-        _bytes[size] = 0;
-        _size = size;
-        if (_depth != -1)
-            _depth++;
-        bumpGeneration();
-        return this;
+        final int save = _size;
+        try {
+            int size = _size;
+            _bytes[size++] = TYPE_BEFORE;
+            _bytes[size] = 0;
+            _size = size;
+            if (_depth != -1)
+                _depth++;
+            bumpGeneration();
+            return this;
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -3496,34 +3558,54 @@
      * operations.
      */
     Key appendAfter() {
-        int size = _size;
-        _bytes[size++] = (byte) TYPE_AFTER;
-        _bytes[size] = 0;
-        _size = size;
-        if (_depth != -1)
-            _depth++;
-        bumpGeneration();
-        return this;
+        final int save = _size;
+        try {
+            int size = _size;
+            _bytes[size++] = (byte) TYPE_AFTER;
+            _bytes[size] = 0;
+            _size = size;
+            if (_depth != -1)
+                _depth++;
+            bumpGeneration();
+            return this;
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private Key appendDate(final Date v) {
-        _bytes[_size++] = (byte) TYPE_DATE;
-        final int size = appendLongInternal(v.getTime());
-        return endSegment(size);
+        final int save = _size;
+        try {
+            _bytes[_size++] = (byte) TYPE_DATE;
+            final int size = appendLongInternal(v.getTime());
+            return endSegment(size);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private Key appendBigInteger(final BigInteger v) {
-        _bytes[_size++] = TYPE_BIG_INTEGER;
-        appendBigInteger(v, 0);
-        endSegment(_size);
-        return this;
+        final int save = _size;
+        try {
+            _bytes[_size++] = TYPE_BIG_INTEGER;
+            appendBigInteger(v, 0);
+            endSegment(_size);
+            return this;
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private Key appendBigDecimal(final BigDecimal v) {
-        _bytes[_size++] = TYPE_BIG_DECIMAL;
-        appendBigInteger(v.unscaledValue(), v.scale());
-        endSegment(_size);
-        return this;
+        final int save = _size;
+        try {
+            _bytes[_size++] = TYPE_BIG_DECIMAL;
+            appendBigInteger(v.unscaledValue(), v.scale());
+            endSegment(_size);
+            return this;
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     /**
@@ -3541,12 +3623,17 @@
      * @return This <code>Key</code>, to permit method call chaining
      */
     public Key appendByteArray(final byte[] bytes, final int offset, final int size) {
-        _bytes[_size++] = TYPE_BYTE_ARRAY;
-        int keySize = _size;
-        System.arraycopy(bytes, offset, _bytes, keySize, size);
-        _size += size;
-        keySize += quoteNulls(keySize, size, false);
-        return endSegment(keySize);
+        final int save = _size;
+        try {
+            _bytes[_size++] = TYPE_BYTE_ARRAY;
+            int keySize = _size;
+            System.arraycopy(bytes, offset, _bytes, keySize, size);
+            _size += size;
+            keySize += quoteNulls(keySize, size, false);
+            return endSegment(keySize);
+        } catch (final ArrayIndexOutOfBoundsException e) {
+            return tooLong(save);
+        }
     }
 
     private int getTypeCode() {
@@ -3622,6 +3709,11 @@
         return this;
     }
 
+    private Key tooLong(final int originalSize) throws KeyTooLongException {
+        _size = originalSize;
+        throw new KeyTooLongException("Maximum size=" + _maxSize + " original size=" + originalSize);
+    }
+
     private Key setRightEdge() {
         _bytes[0] = (byte) 255;
         _size = 1;
 
=== modified file 'src/main/java/com/persistit/Persistit.java'
--- src/main/java/com/persistit/Persistit.java	2013-03-13 15:55:19 +0000
+++ src/main/java/com/persistit/Persistit.java	2013-03-20 19:33:22 +0000
@@ -1691,7 +1691,7 @@
         releaseAllResources();
     }
 
-    private void closeZombieTransactions(boolean removeAllSessions) {
+    private void closeZombieTransactions(final boolean removeAllSessions) {
         final Set<SessionId> sessionIds;
         synchronized (_transactionSessionMap) {
             sessionIds = new HashSet<SessionId>(_transactionSessionMap.keySet());
 
=== modified file 'src/main/java/com/persistit/Transaction.java'
--- src/main/java/com/persistit/Transaction.java	2013-03-13 16:27:04 +0000
+++ src/main/java/com/persistit/Transaction.java	2013-03-20 19:33:22 +0000
@@ -685,7 +685,7 @@
             }
             try {
                 pruneLockPages();
-            } catch (Exception e) {
+            } catch (final Exception e) {
                 _persistit.getLogBase().pruneException.log(e, "locks");
             }
             _transactionStatus = null;
 
=== added file 'src/main/java/com/persistit/exception/KeyTooLongException.java'
--- src/main/java/com/persistit/exception/KeyTooLongException.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/persistit/exception/KeyTooLongException.java	2013-03-20 19:33:22 +0000
@@ -0,0 +1,39 @@
+/**
+ * Copyright © 2005-2012 Akiban Technologies, Inc.  All rights reserved.
+ * 
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Public License v1.0 which
+ * accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * This program may also be available under different license terms.
+ * For more information, see www.akiban.com or contact [e-mail].
+ * 
+ * Contributors:
+ * Akiban Technologies, Inc.
+ */
+
+package com.persistit.exception;
+
+/**
+ * Thrown by {@link com.persistit.Key} when an attempt to append a key segment
+ * exceeds the maximum size of the key.
+ * 
+ * @version 1.0
+ */
+public class KeyTooLongException extends RuntimeException {
+
+    private static final long serialVersionUID = -4657691754665822537L;
+
+    public KeyTooLongException() {
+        super();
+    }
+
+    public KeyTooLongException(final String msg) {
+        super(msg);
+    }
+
+    public KeyTooLongException(final Throwable t) {
+        super(t);
+    }
+}
 
=== modified file 'src/test/java/com/persistit/TransactionAbandonedTest.java'
--- src/test/java/com/persistit/TransactionAbandonedTest.java	2013-02-20 17:13:17 +0000
+++ src/test/java/com/persistit/TransactionAbandonedTest.java	2013-03-20 19:33:22 +0000
@@ -15,13 +15,14 @@
 
 package com.persistit;
 
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Before;
+import org.junit.Test;
+
 import com.persistit.exception.PersistitException;
 import com.persistit.unit.ConcurrentUtil;
 import com.persistit.unit.UnitTestProperties;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
 
 /**
  * <p>
@@ -51,7 +52,7 @@
         private final boolean doRead;
         private final boolean doWrite;
 
-        public TxnAbandoner(Persistit persistit, boolean doRead, boolean doWrite) {
+        public TxnAbandoner(final Persistit persistit, final boolean doRead, final boolean doWrite) {
             this.persistit = persistit;
             this.doRead = doRead;
             this.doWrite = doWrite;
@@ -59,7 +60,7 @@
 
         @Override
         public void run() throws PersistitException {
-            Transaction txn = persistit.getTransaction();
+            final Transaction txn = persistit.getTransaction();
             txn.begin();
             if (doRead) {
                 assertEquals("Traverse count", KEY_RANGE, scanAndCount(getExchange(persistit)));
@@ -70,18 +71,19 @@
         }
     }
 
-    private static Exchange getExchange(Persistit persistit) throws PersistitException {
+    private static Exchange getExchange(final Persistit persistit) throws PersistitException {
         return persistit.getExchange(UnitTestProperties.VOLUME_NAME, TREE, true);
     }
 
-    private static void loadData(Persistit persistit, int keyOffset, int count) throws PersistitException {
-        Exchange ex = getExchange(persistit);
+    private static void loadData(final Persistit persistit, final int keyOffset, final int count)
+            throws PersistitException {
+        final Exchange ex = getExchange(persistit);
         for (int i = 0; i < count; ++i) {
             ex.clear().append(keyOffset + i).store();
         }
     }
 
-    private static int scanAndCount(Exchange ex) throws PersistitException {
+    private static int scanAndCount(final Exchange ex) throws PersistitException {
         ex.clear().append(Key.BEFORE);
         int saw = 0;
         while (ex.next()) {
@@ -96,8 +98,8 @@
         loadData(_persistit, KEY_START, KEY_RANGE);
     }
 
-    private void runAndCleanup(String name, boolean doRead, boolean doWrite) {
-        Thread t = ConcurrentUtil.createThread(name, new TxnAbandoner(_persistit, false, false));
+    private void runAndCleanup(final String name, final boolean doRead, final boolean doWrite) {
+        final Thread t = ConcurrentUtil.createThread(name, new TxnAbandoner(_persistit, false, false));
         ConcurrentUtil.startAndJoinAssertSuccess(MAX_TIMEOUT_MS, t);
         // Threw exception before fix
         _persistit.cleanup();
@@ -119,11 +121,11 @@
         runAndCleanup("ReadAndWrite", true, true);
         assertEquals("Traversed after abandoned", KEY_RANGE, scanAndCount(getExchange(_persistit)));
         // Check that the abandoned was pruned
-        CleanupManager cm = _persistit.getCleanupManager();
+        final CleanupManager cm = _persistit.getCleanupManager();
         for (int i = 0; i < 5 && cm.getEnqueuedCount() > 0; ++i) {
             cm.runTask();
         }
-        Exchange rawEx = getExchange(_persistit);
+        final Exchange rawEx = getExchange(_persistit);
         rawEx.ignoreMVCCFetch(true);
         assertEquals("Raw traversed after abandoned", KEY_RANGE, scanAndCount(rawEx));
     }
 
=== modified file 'src/test/java/com/persistit/unit/ExchangeTest.java'
--- src/test/java/com/persistit/unit/ExchangeTest.java	2012-12-24 18:27:56 +0000
+++ src/test/java/com/persistit/unit/ExchangeTest.java	2013-03-20 19:33:22 +0000
@@ -31,7 +31,7 @@
 import com.persistit.PersistitUnitTestCase;
 import com.persistit.Transaction;
 import com.persistit.Volume;
-import com.persistit.exception.ConversionException;
+import com.persistit.exception.KeyTooLongException;
 import com.persistit.exception.PersistitException;
 
 public class ExchangeTest extends PersistitUnitTestCase {
@@ -151,8 +151,8 @@
         randomKey = createString(initialLength);
         try {
             ex.clear().append(randomKey);
-            fail("ConversionException should have been thrown");
-        } catch (final ConversionException expected) {
+            fail("KeyTooLongException should have been thrown");
+        } catch (final KeyTooLongException expected) {
         }
     }
 
 
=== modified file 'src/test/java/com/persistit/unit/KeyTest1.java'
--- src/test/java/com/persistit/unit/KeyTest1.java	2012-11-07 14:03:51 +0000
+++ src/test/java/com/persistit/unit/KeyTest1.java	2013-03-20 19:33:22 +0000
@@ -22,6 +22,7 @@
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.util.Date;
 
 import junit.framework.Assert;
 
@@ -33,8 +34,8 @@
 import com.persistit.PersistitUnitTestCase;
 import com.persistit.TestShim;
 import com.persistit.exception.InvalidKeyException;
+import com.persistit.exception.KeyTooLongException;
 import com.persistit.exception.MissingKeySegmentException;
-import com.persistit.util.Util;
 
 public class KeyTest1 extends PersistitUnitTestCase {
 
@@ -45,22 +46,26 @@
     private double dv1;
     private double dv2;
 
-    long[] TEST_LONGS = { 0, 1, 2, 3, 123, 126, 127, 128, 129, 130, 131, 132, 250, 251, 252, 253, 254, 255, 256, 257,
-            258, 259, 260, 4094, 4095, 4096, 4097, 4098, 16383, 16384, 16385, 0x1FFFFEL, 0x1FFFFFL, 0x200000L,
-            0x3FFFFFFFFFEL, 0x3FFFFFFFFFFL, 0x2000000000000L, 0x1FFFFFFFFFFFFL, 0x1FFFFFFFFFFFEL,
+    private final static long[] TEST_LONGS = { 0, 1, 2, 3, 123, 126, 127, 128, 129, 130, 131, 132, 250, 251, 252, 253,
+            254, 255, 256, 257, 258, 259, 260, 4094, 4095, 4096, 4097, 4098, 16383, 16384, 16385, 0x1FFFFEL, 0x1FFFFFL,
+            0x200000L, 0x3FFFFFFFFFEL, 0x3FFFFFFFFFFL, 0x2000000000000L, 0x1FFFFFFFFFFFFL, 0x1FFFFFFFFFFFEL,
             Integer.MAX_VALUE - 2, Integer.MAX_VALUE - 1, Integer.MAX_VALUE + 0, Integer.MAX_VALUE + 1,
             Integer.MAX_VALUE + 2, Integer.MAX_VALUE + 3, Long.MIN_VALUE, Long.MIN_VALUE + 1, Long.MIN_VALUE + 2,
             Long.MAX_VALUE, Long.MAX_VALUE - 1, Long.MAX_VALUE - 2, };
 
-    float[] TEST_FLOATS = { 0.0F, 1.0F, 2.0F, 12345.0F, 0.0003F, 1.2345E-10F, 1.12345E-20F, 0.0F, -1.0F, -2.0F,
-            -12345.0F, -0.0003F, -1.2345E-10F, -1.12345E-20F, Float.MAX_VALUE, Float.MIN_VALUE, Float.MAX_VALUE / 2.0F,
-            Float.MAX_VALUE / 3.0F, Float.MAX_VALUE / 4.0F, Float.MIN_VALUE / 2.0F, Float.MIN_VALUE / 3.0F,
-            Float.MIN_VALUE / 4.0F, Float.NaN, Float.NEGATIVE_INFINITY, Float.POSITIVE_INFINITY };
-
-    double[] TEST_DOUBLES = { 0.0, 1.0, 2.0, 12345.0, 0.0003, 1.2345E-10, 1.12345E-20, 0.0, -1.0, -2.0, -12345.0,
-            -0.0003, -1.2345E-10, -1.12345E-20, Double.MAX_VALUE, Double.MIN_VALUE, Double.MAX_VALUE / 2.0,
-            Double.MAX_VALUE / 3.0, Double.MAX_VALUE / 4.0, Double.MIN_VALUE / 2.0, Double.MIN_VALUE / 3.0,
-            Double.MIN_VALUE / 4.0, Double.NaN, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY };
+    private final static float[] TEST_FLOATS = { 0.0F, 1.0F, 2.0F, 12345.0F, 0.0003F, 1.2345E-10F, 1.12345E-20F, 0.0F,
+            -1.0F, -2.0F, -12345.0F, -0.0003F, -1.2345E-10F, -1.12345E-20F, Float.MAX_VALUE, Float.MIN_VALUE,
+            Float.MAX_VALUE / 2.0F, Float.MAX_VALUE / 3.0F, Float.MAX_VALUE / 4.0F, Float.MIN_VALUE / 2.0F,
+            Float.MIN_VALUE / 3.0F, Float.MIN_VALUE / 4.0F, Float.NaN, Float.NEGATIVE_INFINITY, Float.POSITIVE_INFINITY };
+
+    private final static double[] TEST_DOUBLES = { 0.0, 1.0, 2.0, 12345.0, 0.0003, 1.2345E-10, 1.12345E-20, 0.0, -1.0,
+            -2.0, -12345.0, -0.0003, -1.2345E-10, -1.12345E-20, Double.MAX_VALUE, Double.MIN_VALUE,
+            Double.MAX_VALUE / 2.0, Double.MAX_VALUE / 3.0, Double.MAX_VALUE / 4.0, Double.MIN_VALUE / 2.0,
+            Double.MIN_VALUE / 3.0, Double.MIN_VALUE / 4.0, Double.NaN, Double.NEGATIVE_INFINITY,
+            Double.POSITIVE_INFINITY };
+
+    private final static Object[] TEST_OBJECTS = new Object[] { null, true, false, (byte) 1, (char) 2, (short) 3, 4,
+            (long) 5, 6.0f, 7.0d, "This is a String", new Date(), new BigInteger("1"), new BigDecimal("2.2") };
 
     @Test
     public void test1() {
@@ -1013,6 +1018,24 @@
         assertEquals("Key value incorrect", "{1,2,3,\"abc\",\"abc\"}", key1.toString());
     }
 
+    @Test
+    public void testKeyTooLong() throws Exception {
+        for (int maxSize = 1; maxSize < 99; maxSize++) {
+            mainLoop: for (final Object value : TEST_OBJECTS) {
+                final Key key = new Key(_persistit, maxSize);
+                // Every appended value consumes at least 2 bytes
+                for (int i = 0; i < 50; i++) {
+                    try {
+                        key.append(value);
+                    } catch (final KeyTooLongException e) {
+                        continue mainLoop;
+                    }
+                }
+                fail("Should have thrown a KeyTooLongException for maxSize=" + maxSize + " and value " + value);
+            }
+        }
+    }
+
     private static boolean doubleEquals(final double f1, final double f2) {
         if (Double.isNaN(f1)) {
             return Double.isNaN(f2);
@@ -1023,45 +1046,6 @@
         return f1 == f2;
     }
 
-    @Override
-    public void runAllTests() throws Exception {
-        test1();
-        test2();
-        test3();
-        test4();
-        test5();
-        test6();
-        test7();
-        test8();
-        test9();
-        test10();
-        test11();
-        test12();
-        test13();
-        test14();
-    }
-
-    private String floatBits(final float v) {
-        final int bits = Float.floatToIntBits(v);
-        final StringBuilder sb = new StringBuilder();
-        Util.hex(sb, bits, 8);
-        return sb.toString();
-    }
-
-    private String doubleBits(final double v) {
-        final long bits = Double.doubleToLongBits(v);
-        final StringBuilder sb = new StringBuilder();
-        Util.hex(sb, bits, 16);
-        return sb.toString();
-    }
-
-    private void debug(final boolean condition) {
-        if (!condition) {
-            return;
-        }
-        return; // <-- breakpoint here
-    }
-
     private Key newKey() {
         return new Key(_persistit);
     }