tree 8a23dce9630e7e7caa6e7db31ddaf33448416365
parent da8d1582cc4cff22c23d80dc117ea4f076ca3827
author ajo thomas <ajothomas@users.noreply.github.com> 1617149533 -0700
committer GitHub <noreply@github.com> 1617149533 -0700
gpgsig -----BEGIN PGP SIGNATURE-----
 
 wsBcBAABCAAQBQJgY75dCRBK7hj4Ov3rIwAAdHIIAIepI2LnG3zeF4yAwDjw0AP0
 4OAwVEdw2oIw0pNOLCL8a3VQIIUJBKSPU2f+HaZ/q2mztfaHipyqOiQ62o6f6+Dg
 XLjW/a/rIAEBArzRmMqD+s+sm0hHGJ+dYo0keITL6zESpNofkYS2YytmRP40cRrP
 BRTmivLL2y2ddDQNM5v33TBXzOM0Gs5QRq9Aees+5c+epKUermuCeLyLvTGPZ9Nm
 P2PUEAB8XEZK9vsyKeRwvb2X64FHQZhPNZLpZxnvd2TZKy2q1uQrrUWvkYLaXpH1
 thq8ZBTFrykU8PosJdD86uuji5D45rF6c1yz5aNjvIG769Mb5MhHgQ9kOIJlj9s=
 =8ALZ
 -----END PGP SIGNATURE-----
 

Changed StreamOperatorTask to throw appropriate Exception when InputOperator is not found for a SystemStream in the OperatorGraph (#1483)

Symptom:
Samza job fails with a TaskCallbackTimeoutException if InputOperator is not found in OperatorGraph for a SystemStream

Cause:
Currently, the pipeline silently fails when there is no InputOperator for a SystemStream in the OperatorGraph as the code doesn't handle such cases. The pipeline just silently fails and the job would throw an exception due to sync task callback timeout (TaskCallbackTimeoutException).
One of the scenarios that causes this is due to non-determinism in the plan seen by the containers and the AM. It can happen in multiple scenarios where the transform/operators have non-determinism in generating their names. e.g. we noticed beam transform appended hashcode.

Changes:
Call failure method of the TaskCallback with a SamzaException with the appropriate exception message.