SAMZA-2176: Ignore the configuration keys with serialized null values from the coordinator stream.

[shanthoosh]

Samza stores the metadata of the job into a log-compacted kafka topic called coordinator stream.

Due to some operational issue, we converted serialized null value into bytes and stored them as values for some keys in the coordinator stream. These keys will not be log-compacted by kafka, since they're not actually null messages.

This patch ignores the configurations from coordinator stream with serialized-null values and buffers only keys with non-null values in memory in ApplicationMaster.

Verified the changes in this patch by testing with a samza-hello-samza job by injecting samza-reserved configuration keys with serialized null values into the coordinator stream. Here is the entire local testing log indicating successful verification.

[shanthoosh]

@cameronlee314
In addition to introducing non-null checks for configuration, I also added non-null checks for all types of metadata in different utility class classes viz TaskAssignmentManager, LocalityManager, TaskPartitionAssignmentManager etc.

In the future, when someone by mistake writes serialized null(random-empty) for these other message types in coordinator stream, these null-checks would guard the application from failing. Please take a look when you get a chance.

[rmatharu-zz]

Thanks for the changes.
Minor comments/consolidation possible

[shanthoosh]

Thanks for the review.

[cameronlee314]

For the non-config cases, did the old 1.0 logic do the null checks?

[shanthoosh]

@cameronlee314

For non-config cases, old 1.0 logic did not do the null checks.

This whole PR was necessary because coordinator stream was corrupted with serialized null values for config-type keys. The same serialized-null values can be published to coordinator stream for any type of keys(container-host, task-partition) in the future.

I'm adding this null-checks as a safe-guard to prevent against it.

[cameronlee314]

@cameronlee314

For non-config cases, old 1.0 logic did not do the null checks.

This whole PR was necessary because coordinator stream was corrupted with serialized null values for config-type keys. The same serialized-null values can be published to coordinator stream for any type of keys(container-host, task-partition) in the future.

I'm adding this null-checks as a safe-guard to prevent against it.

I suggest doing these two changes in separate PRs. First PR to just fix the config case to bring it back to Samza 1.0 functionality. Second PR to add the safe guards. Then we can manage the changes separately (e.g. if it turns out that the extra non-config safe guards are undesirable, we don't want to roll back the config changes).
One of the goals of this change was to get back to Samza 1.0 functionality, but this PR doesn't quite do that.

[shanthoosh]

if it turns out that the extra non-config safe guards are undesirable, we don't want to roll back the config changes

Removed the null-checks for non-config key types from this PR.

[cameronlee314]

LGTM