shanthoosh commented on a change in pull request #918: SAMZA-2094: Implement the StartpointVisitor for the KafkaSystemConsumer. URL: https://github.com/apache/samza/pull/918#discussion_r266258426
########## File path: samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java ########## @@ -60,16 +74,22 @@ private final Config config; private final boolean fetchThresholdBytesEnabled; private final KafkaSystemConsumerMetrics metrics; + private final KafkaStartpointRegistrationHandler kafkaStartpointRegistrationHandler; // This sink is used to transfer the messages from the proxy/consumer to the BlockingEnvelopeMap. final KafkaConsumerMessageSink messageSink; // This proxy contains a separate thread, which reads kafka messages (with consumer.poll()) and populates // BlockingEnvelopMap's buffers. - final private KafkaConsumerProxy proxy; + @VisibleForTesting + KafkaConsumerProxy proxy; // keep registration data until the start - mapping between registered SSPs and topicPartitions, and their offsets final Map<TopicPartition, String> topicPartitionsToOffset = new HashMap<>(); Review comment: 1. Created PR: 954 which addresses the comment of moving the refactor into separate PR. 2. Added the rationale in description of PR: 954 as to why this refactor would not break any existing functionality. * `CoordinatorStreamSystemConsumer`: Reads the coordinator stream from oldest offset. Please check here(https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemConsumer.java#L128) for more details. * `CoordinatorStreamStore`: Reads the coordinator stream from oldest offset. Please check [here](https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java#L183) for more details. * `ContainerStorageManager`: Chooses the lowest offset before registering with consumer. [Relevant code](https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java#L68) . Please let me know if there're any other questions. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services