[SAMZA-2071] Skip StartpointManager instantiation for PassThroughCoordinator.

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.1
• Fix Version/s: 1.1
• Assignee: Shanthoosh Venkataraman
• Reporter: Shanthoosh Venkataraman
• Created: 15/Jan/19 23:51
• Updated: 07/Mar/19 01:51
• Resolved: 07/Feb/19 15:57

Description

Samza support PassThroughCoordinator where the coordination between processors and checkpoints are expected to be managed by the external service. Coordinator stream is used in samza to persist the metadata related to the coordiantion and a execution. JobCoordinator is not applicable for the PassthroughCoordinator case in samza.

SAMZA-1985 added a startpoint manager implementation and by default used coordinator stream as the startpoint metadata store. After this, running PassthroughCoordinator locally require coordinator stream and the associated coordinator configurations. If it is not defined, it spams the log with the following exception.

org.apache.samza.config.ConfigException: Missing job.coordinator.system configuration. Cannot proceed with job execution. at org.apache.samza.config.JobConfig.getCoordinatorSystemName(JobConfig.scala:145)
at org.apache.samza.util.CoordinatorStreamUtil$.getCoordinatorSystemStream(CoordinatorStreamUtil.scala:56)
at org.apache.samza.util.CoordinatorStreamUtil.getCoordinatorSystemStream(CoordinatorStreamUtil.scala)
at org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore.<init>(CoordinatorStreamStore.java:81)
at org.apache.samza.coordinator.metadatastore.CoordinatorStreamMetadataStoreFactory.getMetadataStore(CoordinatorStreamMetadataStoreFactory.java:34)
at org.apache.samza.startpoint.StartpointManager.<init>(StartpointManager.java:77)
at org.apache.samza.container.SamzaContainer$.apply(SamzaContainer.scala:428)
at org.apache.samza.container.SamzaContainer.apply(SamzaContainer.scala)
at org.apache.samza.processor.StreamProcessor.createSamzaContainer(StreamProcessor.java:333)
at org.apache.samza.processor.StreamProcessor$1.onNewJobModel(StreamProcessor.java:405)
at org.apache.samza.standalone.PassthroughJobCoordinator.start(PassthroughJobCoordinator.java:103)
at org.apache.samza.processor.StreamProcessor.start(StreamProcessor.java:266)
at java.util.concurrent.ConcurrentHashMap$KeySetView.forEach(ConcurrentHashMap.java:4649)
at org.apache.samza.runtime.LocalApplicationRunner.run(LocalApplicationRunner.java:130)
at org.apache.samza.runtime.ApplicationRunner.run(ApplicationRunner.java:40)
at org.apache.beam.runners.samza.SamzaRunner.run(SamzaRunner.java:110)
at org.apache.beam.runners.samza.TestSamzaRunner.run(TestSamzaRunner.java:75)
at org.apache.beam.runners.samza.runtime.SamzaStoreStateInternalsTest.testMapStateIterator(SamzaStoreStateInternalsTest.java:132)

This hampers local mode development and testing of samza-beam jobs.

Issue Links

• links to: GitHub Pull Request #933