// TODO(hbs): allow setting of writeBufferSize
@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);
          }
          }
        };
      }
TDeserializer deserializer = new TDeserializer(new TCompactProtocol.Factory());
KafkaOffsetCounters counters = pool.getCounters();
// TODO(hbs): allow setting of writeBufferSize
try {
while (iter.hasNext()) {
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());
@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;