Re: Review Request 37817: SAMZA-619 - Modify SamzaAppMaster to enable host-affinity

[Navina Ramesh <nr...@linkedin.com>]

> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-core/src/main/java/org/apache/samza/container/LocalityManager.java, line 82
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055358#file1055358line82>
> >
> >     What's the reason to have both hostname and hostip here? Some javadoc would be nice

Actually, when I made the change to write the container-to-host mapping to the coordinator stream, I used the hostIp (in SAMZA-618). When I started working on host-affinity, I realized that yarn returns the container with metinfo about the host. It contains the hostname rather than the IP. 

I think we can only hostName. I doubt if we are using hostIp anywhere else.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/config/YarnConfig.java, line 51
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055363#file1055363line51>
> >
> >     Just curious: why are we choosing 3.6s as the sleep time here? Is it from experience?

Not really. I just chose those as defaults. We should be able to configure each of them. I will add doc in the configuration table for this. I am not sure what the right defaults are without experimenting with jobs with varying configs.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/config/YarnConfig.java, line 99
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055363#file1055363line99>
> >
> >     Do we need a default value here as well? What's the right behavior if this config variable is not configured? Do we always default to whatever the default Java at the AM machine? Sounded a bit of concern for me.

No. In this case, the null check happens at the caller (See Line 69 in my changes to YarnJob.scala). This was the behavior when this class was in scala and I just translated the same to Java. 

If the option is not configured, it defaults to whatever is defined as JAVA_HOME in the AM machine. Why is this a concern? I think Yarn cluster installation requires a JAVA_HOME to be defined.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/config/YarnConfig.java, line 24
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055363#file1055363line24>
> >
> >     nit: would be nice to add javadoc here.

Ok. I have copied the javadoc from the configuration table.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 32
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line32>
> >
> >     nit: javadoc.

I added a brief intro to this class. I have explained the details in the javadocs for the allocator threads.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 37
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line37>
> >
> >     For better code re-use and readability, I think that it might be worth thinking of creating two derived classes, instead of using a boolean flag in this class?

Hmm.. That was my initial thought. Let me try to make the classes derived and see if it looks better.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 52
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line52>
> >
> >     It would be good to document the overall threading model here. It seems that the following are the main threads involved (correct me if I am wrong):
> >     - The main thread that drive the AM to send out container requests to RM
> >     - The allocator thread here to assign the allocated containers to pending requests
> >     - The callback handler thread that receives the responses from RM and populate the allocated containers collection in containerRequestState
> >     - And the SamzaTaskManager handler thread that handles container failures and re-request the containers from RM
> >     
> >     It would be nice to document the above a little bit here s.t. we have a clear picture of which data structure will be shared among which threads.

Yep. You got it right!

I will document the various threads involved in the SamzaTaskManager class. It is going to be a little tough to explain which datastructures are shared among threads. I have already mentioned them in the ContainerRequestState class.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 71
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line71>
> >
> >     Quick question: it seems the code here tries to take the available containers at the beginning of the loop and looping through the pending requests for allocation. What if the callback handler tries to add to the availableContainers at the same time? Is the List<Container> thread safe? I think the code here still works due to the producer/consumer model on the same structure here, as long as the remove/add of the first element in the List<> does not collide w/ each other.

Even if the callback handler thread adds to availableContainers at the same time, it should not cause any problem because adding containers (in addContainer) and removing containers from the list are both within synchronized blocks. 
Now, it could happen that after we read an empty list of allocated containers on the host, the callback thread adds a container. Even though, we may not enter the while-loop for assignment, it will be considered once the allocator thread wakes up after ALLOCATOR_SLEEP_TIME ms.

Add & remove are synchronized. I don't think it matters whether we synchronize on read for List<Container>. Let me know if you think it is essential here.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 80
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line80>
> >
> >     Would it be better to wrap this together w/ containerRequestState.updateStateAfterAssignment()? If I understand correctly, this step is to make sure that the request queue in the containerRequestState is in sync w/ the buffered requests in amClient. I think that containerRequestState is also updated when a request is send async via amClient. It seems to me that they would be better paired up in two API functions.

Are you suggesting I move the "removeContainerRequest" line within the synchronized block? We are not exactly trying to validate that the request queue in the containerRequestState is in sync with the buffer in the amClient. It doesn't affect the "State" that we maintain. So. I decided to keep it out of the synchronized block. Either way, I don't think it makes a lot of difference. It might make it harder to unit test, though. :)


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 96
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line96>
> >
> >     At this moment, are we truly running out of containers? Or simply need to wait for more containers from the RM? I think that we are not sure here. Would it better to log info "Waiting %s more seconds for more containers to be allocated"?

Agreed. I will change it!


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 138
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line138>
> >
> >     It seems that we can have a common thread class within the ContainerAllocator class and two different classes implementing non-host-affinity and host-affinity algorithms here. Worth to think of?

Hmm.. Haven't thought about it because I was worried about messing up the request states. It will be easier to have 2 derived classes - one for host-affinity and one for non-host-affinity. The common functionalities can be in the parent class. Let me think about this common thread class implementation.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 301
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line301>
> >
> >     If amClient.addContainerRequest() is in the same sync block w/ containerRequestState.updateRequestState() here, I would think that amClient.removeContainerRequest() should be in the same sync block w/ containerRequestState.updateRequestState in the allocator threads too.

Yeah.. That's a reasonable argument.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java, line 314
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055364#file1055364line314>
> >
> >     Just a thought, maybe directly declaring ContainerRequestState.addContainer() to be a synchronized method if it always requires global lock on containerRequestState?

Sure. Making it synchronized should work. However, it will end-up with a some methods being synchronized and some not in ContainerRequestState. That is why I decided to leave the synchronized blocks. 

I don't have a strong opinion on this. I can make it synchornized if you think it makes the code better!


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerRequestState.java, line 34
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055366#file1055366line34>
> >
> >     This is defined redundantly in both ContainerAllocator and here. It would be good to just keep one.

Ok. Makes sense.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerRequestState.java, line 57
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055366#file1055366line57>
> >
> >     I see the opportunity to remove this conditional flag and split this state into two derived classes as well. Thoughts?

Ah.. Ok. For now, I will try to refactor the allocator code with threads :) I will keep this opportunity reserved as a TODO in the code.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerRequestState.java, line 112
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055366#file1055366line112>
> >
> >     So, here to make sure that the request count and the allocated containers count are not changed in the comparison, access of the two variables and update of the two variables need to be atomic. Is it guaranteed by the global lock on the state?
> >     
> >     Or, I guess the worst result from here is that the requestCountOnThisHost has not been updated while the container is allocated. Hence, one allocated container goes to ANY_HOST and missed the chance of being allocated to the preferred host? We may live w/ this if we want to avoid locking here for performance. It would be good to make a note here.

Yeah. The global local on the state object shoud take care of this. I don't think we need to do any kind of locking here. It will lead to double locking issues.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerRequestState.java, line 136
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055366#file1055366line136>
> >
> >     It would good to comment on under which lock this method should be called.

I made this a private method to the class (will be available in the next RB patch) because it is just a helper method. Not intended for any client to directly call it.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/main/scala/org/apache/samza/job/yarn/ContainerUtil.scala, line 19
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055372#file1055372line19>
> >
> >     Any reason this class is still in scala? If it is new/refactored, would it make more sense to change it to java?

:) At one point, I just got tired of moving things to java :P It is hard to find equivalents of certain scala aspects in java. 

If this class doesn't use any functions as parameters, I will move it to java. Otherwise, I prefer to keep it this way for now.


> On Aug. 31, 2015, 9:02 p.m., Yi Pan (Data Infrastructure) wrote:
> > samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala, line 174
> > <https://reviews.apache.org/r/37817/diff/1/?file=1055384#file1055384line174>
> >
> >     I thought that the purpose of these tests are authorganal to the change in host-affinity and would need to be kept here?

So, I nuked the SamzaAppMasterTaskManager.scala class. Part of the operations of that class is now in SamzaTaskManager.java and the rest in ContainerAllocator.java. 
I have added unit tests for each of those classes covering all relevant cases from TestSamzaAppMasterTaskManager.scala. I will update this patch with those changes.


- Navina


-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/37817/#review97059
-----------------------------------------------------------


On Aug. 26, 2015, 10:14 p.m., Navina Ramesh wrote:
> 
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/37817/
> -----------------------------------------------------------
> 
> (Updated Aug. 26, 2015, 10:14 p.m.)
> 
> 
> Review request for samza, Yan Fang, Chinmay Soman, Chris Riccomini, and Yi Pan (Data Infrastructure).
> 
> 
> Bugs: SAMZA-619
>     https://issues.apache.org/jira/browse/SAMZA-619
> 
> 
> Repository: samza
> 
> 
> Description
> -------
> 
> This a major change to how we request and assign containers to resources. It uses a threaded model for request and allocation. More comments in the javadoc. 
> 
> Major changes to note:
> 1. Changed yarn dependency to 2.6
> 2. Moved YarnConfig to java config class
> 2. Removed SamzaAppMasterTaskManager
> 3. SamzaAppState replaces SamzaAppMasterState
> 4. SamzaTaskManager is very similar to SamzaAppMasterTaskManager, except that it delegates onContainerAllocated and requestContainers to the thread in ContainerAllocator 
> 5. Removed state.unclaimedContainers
> 6. Allocator thread sleep time and request timeout is configurable
> 7. host-affinity can be enabled by using the config "yarn.samza.host-affinity.enabled". Host affinity is guaranteed to work with FairScheduler that has continuous scheduling enabled. Details on this config can be found at SAMZA-617 [design](https://issues.apache.org/jira/secure/attachment/12726945/DESIGN-SAMZA-617-2.pdf).
> 
> Pending items:
> 1. Adding unit tests for AppMaster, TaskManager, ContainerAllocator
> 2. Update config documentation 
> 3. Update web-site with info on this feature
> 
> 
> Diffs
> -----
> 
>   build.gradle 3a7fabcd4f4e5c8db950c3a33bba33618c5565b4 
>   gradle/dependency-versions.gradle 36d564b6ca895f042ee4802643e49180f4947b62 
>   samza-core/src/main/java/org/apache/samza/container/LocalityManager.java c567bf4a1747a7a6aff23be74cf25b1a44e57b42 
>   samza-core/src/main/java/org/apache/samza/coordinator/stream/messages/SetContainerHostMapping.java 5455881e2a4a9c865137f2708f655641b93c2bfb 
>   samza-core/src/main/java/org/apache/samza/job/model/JobModel.java 7b592740c082225fc217a24188bc874f883db63b 
>   samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 61e228b1ff8619da4ad2e11b15a5afb7999c4996 
>   samza-core/src/main/scala/org/apache/samza/util/Util.scala c8438bd6ed01a95860867505d84c04c493a5a197 
>   samza-yarn/src/main/java/org/apache/samza/config/YarnConfig.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerAllocator.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerFailure.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/ContainerRequestState.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/SamzaAppState.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/SamzaContainerRequest.java PRE-CREATION 
>   samza-yarn/src/main/java/org/apache/samza/job/yarn/SamzaTaskManager.java PRE-CREATION 
>   samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml ce2145adde09b0e8097a1c711bc296ee20392e63 
>   samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala 1f51551893c42bb13d7941c8b0e5594ac7f42585 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/ContainerUtil.scala PRE-CREATION 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMaster.scala af42c6a6636953a95f79837fe372e0dbd735df70 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterLifecycle.scala d475c4c566f17f88a04db5fbc84cc2e27eb333d2 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala 03acfe1bbbabf8f54be9f36fdae785476da45135 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterService.scala 060538623e4d67b986bc635518e7fe8ebdde9e24 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterState.scala f667c83a7fff43a5efdc64cde019b2cf35f38cb9 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterTaskManager.scala 1743c8611a94fa1c7f7dafd8ff8c713039f3df8c 
>   samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala 8dd70c9977473e083e463d01b049c40e15b21f4a 
>   samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala 09f4dc32a4b18aeb3accb856c360ea2f95c82673 
>   samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala 7fd51221149ba59e7ea1bc0b51d58726ec17a7d7 
>   samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterLifecycle.scala df5992e659302d2918c4e2c30b6122ed51ab9fe8 
>   samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterService.scala 6f4bfaf916d687426f746fd3b09cd56490bb500e 
>   samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 2eec65f02826de40493925c08ff344a8cc4feecb 
> 
> Diff: https://reviews.apache.org/r/37817/diff/
> 
> 
> Testing
> -------
> 
> Tested with hello-samza locally. Tested with a sample job on a 3 node Yarn cluster. Works as expected.
> 
> 
> Thanks,
> 
> Navina Ramesh
> 
>