KafkaSystemConsumer extends BlockingEnvelopeMap implements SystemConsumer { void startConsumer() { if (topicPartitionsToOffset.size() <= 0) { LOG.error("{}: Consumer is not subscribed to any SSPs", this); } topicPartitionsToOffset.forEach((topicPartition, startingOffsetString) -> { long startingOffset = Long.valueOf(startingOffsetString); try { synchronized (kafkaConsumer) { kafkaConsumer.seek(topicPartition, startingOffset); } } catch (Exception e) { String msg = String.format("%s: Got Exception while seeking to %s for partition %s", this, startingOffsetString, topicPartition); LOG.error(msg, e); throw new SamzaException(msg, e); } LOG.info("{}: Changing consumer's starting offset for partition {} to {}", this, topicPartition, startingOffsetString); proxy.addTopicPartition(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), startingOffset); }); if (proxy != null && !proxy.isRunning()) { LOG.info("{}: Starting proxy {}", this, proxy); proxy.start(); } } KafkaSystemConsumer(Consumer<K, V> kafkaConsumer, String systemName, Config config, String clientId,
      KafkaConsumerProxyFactory<K, V> kafkaConsumerProxyFactory, KafkaSystemConsumerMetrics metrics, Clock clock); static KafkaConsumer<K, V> createKafkaConsumerImpl(String systemName, HashMap<String, Object> kafkaConsumerConfig); @Override void start(); @Override void setFailureCause(Throwable throwable); @Override void stop(); @Override void register(SystemStreamPartition systemStreamPartition, String offset); @Override String toString(); @Override Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(
      Set<SystemStreamPartition> systemStreamPartitions, long timeout); String getSystemName(); KafkaConsumerMessageSink getMessageSink();  }
@Test public void testStartConsumer() { final Consumer consumer = Mockito.mock(Consumer.class); final KafkaConsumerProxyFactory kafkaConsumerProxyFactory = Mockito.mock(KafkaConsumerProxyFactory.class); final KafkaSystemConsumerMetrics kafkaSystemConsumerMetrics = new KafkaSystemConsumerMetrics(TEST_SYSTEM, new NoOpMetricsRegistry()); final SystemStreamPartition testSystemStreamPartition1 = new SystemStreamPartition(TEST_SYSTEM, TEST_STREAM, new Partition(0)); final SystemStreamPartition testSystemStreamPartition2 = new SystemStreamPartition(TEST_SYSTEM, TEST_STREAM, new Partition(1)); final String testOffset = "1"; final KafkaConsumerProxy kafkaConsumerProxy = Mockito.mock(KafkaConsumerProxy.class); Mockito.when(kafkaConsumerProxyFactory.create(Mockito.anyObject())).thenReturn(kafkaConsumerProxy); Mockito.doNothing().when(consumer).seek(new TopicPartition(TEST_STREAM, 0), 1); Mockito.doNothing().when(consumer).seek(new TopicPartition(TEST_STREAM, 1), 1); KafkaSystemConsumer kafkaSystemConsumer = new KafkaSystemConsumer(consumer, TEST_SYSTEM, new MapConfig(), TEST_CLIENT_ID, kafkaConsumerProxyFactory, kafkaSystemConsumerMetrics, new SystemClock()); kafkaSystemConsumer.register(testSystemStreamPartition1, testOffset); kafkaSystemConsumer.register(testSystemStreamPartition2, testOffset); kafkaSystemConsumer.startConsumer(); Mockito.verify(consumer).seek(new TopicPartition(TEST_STREAM, 0), 1); Mockito.verify(consumer).seek(new TopicPartition(TEST_STREAM, 1), 1); Mockito.verify(kafkaConsumerProxy).start(); Mockito.verify(kafkaConsumerProxy).addTopicPartition(testSystemStreamPartition1, Long.valueOf(testOffset)); Mockito.verify(kafkaConsumerProxy).addTopicPartition(testSystemStreamPartition2, Long.valueOf(testOffset)); }