@Override
      public Runnable getConsumer(final KafkaSynchronizedConsumerPool pool, final KafkaStream<byte[], byte[]> stream) {
        return new Runnable() {          
          @Override
          public void run() {
            ConsumerIterator<byte[],byte[]> iter = stream.iterator();
            byte[] sipHashKey = frontend.keystore.getKey(KeyStore.SIPHASH_KAFKA_PLASMA_FRONTEND_IN);
            byte[] aesKey = frontend.keystore.getKey(KeyStore.AES_KAFKA_PLASMA_FRONTEND_IN);
            // Iterate on the messages
            TDeserializer deserializer = new TDeserializer(new TCompactProtocol.Factory());
            KafkaOffsetCounters counters = pool.getCounters();
            // TODO(hbs): allow setting of writeBufferSize
            try {
            while (iter.hasNext()) {
              //
              // Since the cal to 'next' may block, we need to first
              // check that there is a message available
              //
              boolean nonEmpty = iter.nonEmpty();
              if (nonEmpty) {
                MessageAndMetadata<byte[], byte[]> msg = iter.next();
                counters.count(msg.partition(), msg.offset());
                byte[] data = msg.message();
                Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_MESSAGES, Sensision.EMPTY_LABELS, 1);
                Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_BYTES, Sensision.EMPTY_LABELS, data.length);
                if (null != sipHashKey) {
                  data = CryptoUtils.removeMAC(sipHashKey, data);
                }
                // Skip data whose MAC was not verified successfully
                if (null == data) {
                  Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_INVALIDMACS, Sensision.EMPTY_LABELS, 1);
                  continue;
                }
                // Unwrap data if need be
                if (null != aesKey) {
                  data = CryptoUtils.unwrap(aesKey, data);
                }
                // Skip data that was not unwrapped successfuly
                if (null == data) {
                  Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_INVALIDCIPHERS, Sensision.EMPTY_LABELS, 1);
                  continue;
                }
                //
                // Extract KafkaDataMessage
                //
                KafkaDataMessage tmsg = new KafkaDataMessage();
                deserializer.deserialize(tmsg, data);
                switch(tmsg.getType()) {
                  case STORE:
                    GTSEncoder encoder = new GTSEncoder(0L, null, tmsg.getData());
                    encoder.setClassId(tmsg.getClassId());
                    encoder.setLabelsId(tmsg.getLabelsId());
                    frontend.dispatch(encoder);
                    break;
                  case DELETE:
                  case ARCHIVE:
                    break;
                  default:
                    throw new RuntimeException("Invalid message type.");
                }            
              } else {
                // Sleep a tiny while
                try {
                  Thread.sleep(1L);
                } catch (InterruptedException ie) {             
                }
              }          
            }        
          } catch (Throwable t) {
            t.printStackTrace(System.err);
          } finally {
            // Set abort to true in case we exit the 'run' method
            pool.getAbort().set(true);
          }
          }
        };
      }
@Override
      public Runnable getConsumer(final KafkaSynchronizedConsumerPool pool, final KafkaStream<byte[], byte[]> stream) {
        return new Runnable() {          
          @Override
          public void run() {
            ConsumerIterator<byte[],byte[]> iter = stream.iterator();
            byte[] sipHashKey = frontend.keystore.getKey(KeyStore.SIPHASH_KAFKA_PLASMA_FRONTEND_IN);
            byte[] aesKey = frontend.keystore.getKey(KeyStore.AES_KAFKA_PLASMA_FRONTEND_IN);
           
            TDeserializer deserializer = new TDeserializer(new TCompactProtocol.Factory());
            KafkaOffsetCounters counters = pool.getCounters();
           
            try {
            while (iter.hasNext()) {
             
             
             
             
              boolean nonEmpty = iter.nonEmpty();
              if (nonEmpty) {
                MessageAndMetadata<byte[], byte[]> msg = iter.next();
                counters.count(msg.partition(), msg.offset());
                byte[] data = msg.message();
                Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_MESSAGES, Sensision.EMPTY_LABELS, 1);
                Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_BYTES, Sensision.EMPTY_LABELS, data.length);
                if (null != sipHashKey) {
                  data = CryptoUtils.removeMAC(sipHashKey, data);
                }
               
                if (null == data) {
                  Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_INVALIDMACS, Sensision.EMPTY_LABELS, 1);
                  continue;
                }
               
                if (null != aesKey) {
                  data = CryptoUtils.unwrap(aesKey, data);
                }
               
                if (null == data) {
                  Sensision.update(SensisionConstants.SENSISION_CLASS_PLASMA_FRONTEND_KAFKA_INVALIDCIPHERS, Sensision.EMPTY_LABELS, 1);
                  continue;
                }
               
               
               
                KafkaDataMessage tmsg = new KafkaDataMessage();
                deserializer.deserialize(tmsg, data);
                switch(tmsg.getType()) {
                  case STORE:
                    GTSEncoder encoder = new GTSEncoder(0L, null, tmsg.getData());
                    encoder.setClassId(tmsg.getClassId());
                    encoder.setLabelsId(tmsg.getLabelsId());
                    frontend.dispatch(encoder);
                    break;
                  case DELETE:
                  case ARCHIVE:
                    break;
                  default:
                    throw new RuntimeException("Invalid message type.");
                }            
              } else {
               
                try {
                  Thread.sleep(1L);
                } catch (InterruptedException ie) {             
                }
              }          
            }        
          } catch (Throwable t) {
            t.printStackTrace(System.err);
          } finally {
           
            pool.getAbort().set(true);
          }
          }
        };
      }
@override public runnable getconsumer(final kafkasynchronizedconsumerpool pool, final kafkastream<byte[], byte[]> stream) { return new runnable() { @override public void run() { consumeriterator<byte[],byte[]> iter = stream.iterator(); byte[] siphashkey = frontend.keystore.getkey(keystore.siphash_kafka_plasma_frontend_in); byte[] aeskey = frontend.keystore.getkey(keystore.aes_kafka_plasma_frontend_in); tdeserializer deserializer = new tdeserializer(new tcompactprotocol.factory()); kafkaoffsetcounters counters = pool.getcounters(); try { while (iter.hasnext()) { boolean nonempty = iter.nonempty(); if (nonempty) { messageandmetadata<byte[], byte[]> msg = iter.next(); counters.count(msg.partition(), msg.offset()); byte[] data = msg.message(); sensision.update(sensisionconstants.sensision_class_plasma_frontend_kafka_messages, sensision.empty_labels, 1); sensision.update(sensisionconstants.sensision_class_plasma_frontend_kafka_bytes, sensision.empty_labels, data.length); if (null != siphashkey) { data = cryptoutils.removemac(siphashkey, data); } if (null == data) { sensision.update(sensisionconstants.sensision_class_plasma_frontend_kafka_invalidmacs, sensision.empty_labels, 1); continue; } if (null != aeskey) { data = cryptoutils.unwrap(aeskey, data); } if (null == data) { sensision.update(sensisionconstants.sensision_class_plasma_frontend_kafka_invalidciphers, sensision.empty_labels, 1); continue; } kafkadatamessage tmsg = new kafkadatamessage(); deserializer.deserialize(tmsg, data); switch(tmsg.gettype()) { case store: gtsencoder encoder = new gtsencoder(0l, null, tmsg.getdata()); encoder.setclassid(tmsg.getclassid()); encoder.setlabelsid(tmsg.getlabelsid()); frontend.dispatch(encoder); break; case delete: case archive: break; default: throw new runtimeexception("invalid message type."); } } else { try { thread.sleep(1l); } catch (interruptedexception ie) { } } } } catch (throwable t) { t.printstacktrace(system.err); } finally { pool.getabort().set(true); } } }; }