#vulnerable code 
                                @Override
    protected void reduce(TableRowkeyPair key, Iterable<ImmutableBytesWritable> values,
                          Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
            throws IOException, InterruptedException {
        TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
        int tableIndex = tableNames.indexOf(key.getTableName());
        List<Pair<byte[], byte[]>> columns = columnIndexes.get(tableIndex);
        for (ImmutableBytesWritable aggregatedArray : values) {
            DataInputStream input = new DataInputStream(new ByteArrayInputStream(aggregatedArray.get()));
            while (input.available() != 0) {
                int index = WritableUtils.readVInt(input);
                Pair<byte[], byte[]> pair = columns.get(index);
                byte type = input.readByte();
                ImmutableBytesWritable value = null;
                int len = WritableUtils.readVInt(input);
                if (len > 0) {
                    byte[] array = new byte[len];
                    input.read(array);
                    value = new ImmutableBytesWritable(array);
                }
                KeyValue kv;
                KeyValue.Type kvType = KeyValue.Type.codeToType(type);
                switch (kvType) {
                    case Put: // not null value
                        kv = builder.buildPut(key.getRowkey(),
                                new ImmutableBytesWritable(pair.getFirst()),
                                new ImmutableBytesWritable(pair.getSecond()), value);
                        break;
                    case DeleteColumn: // null value
                        kv = builder.buildDeleteColumns(key.getRowkey(),
                                new ImmutableBytesWritable(pair.getFirst()),
                                new ImmutableBytesWritable(pair.getSecond()));
                        break;
                    default:
                        throw new IOException("Unsupported KeyValue type " + kvType);
                }
                map.add(kv);
            }
            KeyValue empty = builder.buildPut(key.getRowkey(),
                    emptyFamilyName.get(tableIndex),
                    QueryConstants.EMPTY_COLUMN_BYTES_PTR, ByteUtil.EMPTY_BYTE_ARRAY_PTR);
            map.add(empty);
            Closeables.closeQuietly(input);
        }
        context.setStatus("Read " + map.getClass());
        int index = 0;
        for (KeyValue kv : map) {
            context.write(key, kv);
            if (++index % 100 == 0) context.setStatus("Wrote " + index);
        }
    } 
                            #location 20 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                                @Override
    protected void reduce(TableRowkeyPair key, Iterable<ImmutableBytesWritable> values,
                          Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
            throws IOException, InterruptedException {
        TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
        ImmutableBytesWritable rowKey = key.getRowkey();
        for (ImmutableBytesWritable aggregatedArray : values) {
            DataInputStream input = new DataInputStream(new ByteArrayInputStream(aggregatedArray.get()));
            while (input.available() != 0) {
                byte type = input.readByte();
                int index = WritableUtils.readVInt(input);
                ImmutableBytesWritable family;
                ImmutableBytesWritable name;
                ImmutableBytesWritable value = QueryConstants.EMPTY_COLUMN_VALUE_BYTES_PTR;
                if (index == -1) {
                    family = emptyFamilyName.get(key.getTableName());
                    name = QueryConstants.EMPTY_COLUMN_BYTES_PTR;
                } else {
                    Pair<byte[], byte[]> pair = columnIndexes.get(index);
                    if(pair.getFirst() != null) {
                        family = new ImmutableBytesWritable(pair.getFirst());
                    } else {
                        family = emptyFamilyName.get(key.getTableName());
                    }
                    name = new ImmutableBytesWritable(pair.getSecond());
                }
                int len = WritableUtils.readVInt(input);
                if (len > 0) {
                    byte[] array = new byte[len];
                    input.read(array);
                    value = new ImmutableBytesWritable(array);
                }
                KeyValue kv;
                KeyValue.Type kvType = KeyValue.Type.codeToType(type);
                switch (kvType) {
                    case Put: // not null value
                        kv = builder.buildPut(key.getRowkey(), family, name, value);
                        break;
                    case DeleteColumn: // null value
                        kv = builder.buildDeleteColumns(key.getRowkey(), family, name);
                        break;
                    default:
                        throw new IOException("Unsupported KeyValue type " + kvType);
                }
                map.add(kv);
            }
            Closeables.closeQuietly(input);
        }
        context.setStatus("Read " + map.getClass());
        int index = 0;
        for (KeyValue kv : map) {
            context.write(key, kv);
            if (++index % 100 == 0) context.setStatus("Wrote " + index);
        }
    }