KafkaSystemConsumer extends BlockingEnvelopeMap implements SystemConsumer { @Override public void start() { if (!started.compareAndSet(false, true)) { LOG.warn("{}: Attempting to start the consumer for the second (or more) time.", this); return; } if (stopped.get()) { LOG.error("{}: Attempting to start a stopped consumer", this); return; } startSubscription(); setFetchThresholds(); startConsumer(); LOG.info("{}: Consumer started", this); } 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 testConfigValidations() { final KafkaSystemConsumer consumer = createConsumer(FETCH_THRESHOLD_MSGS, FETCH_THRESHOLD_BYTES); consumer.start(); }