Major Description
If a StreamTask calls TaskCoordinator.commit, all TaskInstances will commit their SystemProducers, TaskStorageManager, and CheckpointManager. The problem with this is that if you have 400 partitions in a SamzaContainer (for example), and each calls TaskCoordinator.commit once per second, you actually get 400 commits per TaskInstance per second. This is incorrect behavior. We should make TaskInstance commit itself only when its task.commit.ms window has expired, or when TaskCoordinator.commit was called ONLY by its StreamTask partition.