protected boolean hasError(ByteBufferMessageSet messages)
            throws IOException {
        int errorCode = messages.getErrorCode();
        if (errorCode == ErrorMapping.OffsetOutOfRangeCode()) {
            /* offset cannot cross the maximum offset (guaranteed by Kafka protocol).
               Kafka server may delete old files from time to time */
            System.err.println("WARNING: current offset=" + offset + ". It is out of range.");

            if (retry >= MAX_RETRY_TIME)  return true;
            retry++;
            // get the current offset range
            offset =  split.getOffset();
            return false;
        } else if (errorCode == ErrorMapping.InvalidMessageCode()) {
            throw new IOException(split + " current offset=" + offset
                    + " : invalid offset.");
        } else if (errorCode == ErrorMapping.WrongPartitionCode()) {
            throw new IOException(split + " : wrong partition");
        } else if (errorCode != ErrorMapping.NoError()) {
            throw new IOException(split + " current offset=" + offset
                    + " error:" + errorCode);
        } else
            return false;
    }