@Override
        public void run() {
            if (logger.isLoggable(Level.FINEST)) {
                logger.finest("Starting row diff change processor");
            }
            while (true) {
                try {
                    IRow r = queue.take();
                    if (r instanceof PoisonRow) {
                        //stop processing
                        break;
                    }
                    for (DiffListener l : listeners) {
                        try {
                            l.newRow(r);
                        } catch (Throwable ex) {
                            StringBuilder sb = new StringBuilder();
                            sb.append("Error while notifying listener ")
                                    .append(l.getClass().getName())
                                    .append(" with new row notification for row ")
                                    .append(r.toString());
                            logger.log(Level.SEVERE, sb.toString(), ex);
                        }
                    }
                } catch (InterruptedException ex) {
                    Thread.interrupted();
                }
            }
            if (logger.isLoggable(Level.FINEST)) {
                logger.finest("Shutting down row diff change processor.");
            }
        }