[jira] [Commented] (SAMZA-465) Use coordinator stream and eliminate CheckpointManager

["Yan Fang (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SAMZA-465?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14522894#comment-14522894 ] 

Yan Fang commented on SAMZA-465:
--------------------------------

{quote}
1. You are right, the consumer really doesn't do anything in the Container, it's a no-op, I can wipe that out.
{quote}

Yes, I think we should remove it. Because the consumer actually creates a thread that is not being used.

{quote}
 Your understanding is correct, the reason why I left it like this is because, we need to refactor the base class (which does the bootstrapping). I wanted to see one more pattern before we abstract out components, when Navina is working on SAMZA-557, we will need to store host-machine name mapping in the Coordinatorstream as well, probably "HostMappingManager". I'll abstract out the base class which shares the same coordinator stream producer and consumer. Ultimately there will be one base/abstract class which shares the same functionality and same coordinator stream consumer/producer, and all the "managers" will share that. I'll follow up on this clean-up.
{quote}

Ok. That makes sense. Maybe a follow-up ticket to track this?

{quote}
4. initializeJobModel() returns a lamba (Jobcoordinator.scala:184). For every HTTP request, this lamba is invoked, and within the lambda, we refresh the checkpoint.
{quote}

I see. understood. (This part is really confusing me...) 

The reason I got confused is because (1) the name of "initializeJobModel" is misled. The initializeJobModel does too many jobs -- not only "initialize" a JobModel, but also return a "handler" for refreshing (2) the refreshJobModel's doc says "The function reads the latest checkpoint from the underlying coordinator stream and builds a new JobModel", but actually it does more than that -- it updates changelogMap and regrouping and is even used for initializing JobModel (line 191- line 207), not only in "refreshing".

So my suggestion is that, a) if we really want the refreshJobModel to do so many jobs, need to change to a more understandable name and the doc. We should not differentiate "initialize" and "refresh", because they are essentially the same. b) If we really want to only do the "refreshJobModel", can we only update the checkpoint, not include updating changelogPartitionMapping and regrouping? Even though in the future, we may need to include changelogPartition and regrouping in refresh, this will not happen shortly. Because it has a few dependencies: 1) JobCoordinator/AM swap 2) read latest config to get new container count and new other configs 3) new implementation of changelog and state-storage. For now, I think it's good to keep it clean.

Nits in current refreshJobModel method:
* allSystemStreamPartitions (line 219) is not used in this method, should be removed.
* "val containerGrouper = new GroupByContainerCount(containerCount)" will never change because containerCount is passed by the initialization step, not from stream. 
* "changelogPartition" never change either because the groups do not change. They are defined in the initialization step too.

Sorry for being a little picky here. *_*

{quote}
the same lamba is being called for each HTTP request
{quote}

Do we have lots of the HTTP request ? I am only aware of that one request happens when we start a container. What are other requests?

{quote}
 the way the Checkpoint is being refreshed
{quote}

I think we may need to not only refresh checkpoint, but also configs when we consider supporting dynamic changes of the config. Of course, that's in another ticket.

{quote}
we clean up the JobCoordinator/AM swap.
{quote}
which ticket is this? I can not find it.

> Use coordinator stream and eliminate CheckpointManager
> ------------------------------------------------------
>
>                 Key: SAMZA-465
>                 URL: https://issues.apache.org/jira/browse/SAMZA-465
>             Project: Samza
>          Issue Type: Sub-task
>          Components: container, kafka
>    Affects Versions: 0.10.0
>            Reporter: Chris Riccomini
>            Assignee: Naveen
>             Fix For: 0.10.0
>
>         Attachments: 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, 0001-Coordinator-stream.patch, SAMZA-465.patch
>
>
> SAMZA-448 introduced a coordinator stream into Samza, but still kept CheckpointManagers. This means that master currently has two streams, which is totally unnecessary. We should eliminate the CheckpointManager, and use the coordinator stream instead.
> This work will involve modifying the JobCoordinator to continue reading messages from the coordinator stream even after it starts up. This is required because a container may fail and restart. When it does so, the offset that it receives from the JobCoordinator (via the HTTP JSON server) must be up to date.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)