Package com.linkedin.venice.samza

Class VeniceSystemFactory

java.lang.Object
com.linkedin.venice.samza.VeniceSystemFactory
All Implemented Interfaces:
Serializable, org.apache.samza.system.SystemFactory
public class VeniceSystemFactory extends Object implements org.apache.samza.system.SystemFactory, Serializable
Samza jobs talk to either parent or child controller depending on the aggregate mode config. The decision of which controller should be used is made in VeniceSystemFactory. The "Primary Controller" term is used to refer to whichever controller the Samza job should talk to. The primary controller should be: 1. The parent controller when the Venice system is deployed in a multi-colo mode and either: a. PushType is PushType.BATCH or PushType.STREAM_REPROCESSING; or b. @deprecated PushType is PushType.STREAM and the job is configured to write data in AGGREGATE mode 2. The child controller when either: a. The Venice system is deployed in a single-colo mode; or b. The PushType is PushType.STREAM and the job is configured to write data in NON_AGGREGATE mode
See Also:

  • Field Details

    • LEGACY_CHILD_D2_ZK_HOSTS_PROPERTY

      public static final String LEGACY_CHILD_D2_ZK_HOSTS_PROPERTY
      See Also:

    • SYSTEMS_PREFIX

      public static final String SYSTEMS_PREFIX
      See Also:

    • DOT

      public static final String DOT
      See Also:

    • DEPLOYMENT_ID

      public static final String DEPLOYMENT_ID
      See Also:

    • VENICE_PUSH_TYPE

      public static final String VENICE_PUSH_TYPE
      See Also:

    • VENICE_STORE

      public static final String VENICE_STORE
      Venice store name Samza application is going to produce to.
      See Also:

    • VENICE_AGGREGATE

      public static final String VENICE_AGGREGATE
      Whether to leverage Venice aggregation. By default, it is 'false'. When the Samza application decides to leverage Venice aggregation, all the messages will be produced to Venice Parent cluster, otherwise, all the messages will be produced to local cluster.
      See Also:

    • VENICE_CHILD_D2_ZK_HOSTS

      public static final String VENICE_CHILD_D2_ZK_HOSTS
      D2 ZK hosts for Venice Child Cluster.
      See Also:

    • VENICE_CONTROLLER_DISCOVERY_URL

      public static final String VENICE_CONTROLLER_DISCOVERY_URL
      See Also:

    • VENICE_ROUTER_URL

      public static final String VENICE_ROUTER_URL
      See Also:

    • VENICE_PARENT_D2_ZK_HOSTS

      public static final String VENICE_PARENT_D2_ZK_HOSTS
      D2 ZK hosts for Venice Parent Cluster.
      See Also:

    • VENICE_CHILD_CONTROLLER_D2_SERVICE

      public static final String VENICE_CHILD_CONTROLLER_D2_SERVICE
      See Also:

    • VENICE_PARENT_CONTROLLER_D2_SERVICE

      public static final String VENICE_PARENT_CONTROLLER_D2_SERVICE
      See Also:

    • LEGACY_VENICE_CHILD_CONTROLLER_D2_SERVICE

      public static final String LEGACY_VENICE_CHILD_CONTROLLER_D2_SERVICE
      See Also:

    • LEGACY_VENICE_PARENT_CONTROLLER_D2_SERVICE

      public static final String LEGACY_VENICE_PARENT_CONTROLLER_D2_SERVICE
      See Also:

  • Constructor Details

    • VeniceSystemFactory

      public VeniceSystemFactory()

  • Method Details

    • getAdmin

      public org.apache.samza.system.SystemAdmin getAdmin(String systemName, org.apache.samza.config.Config config)
      Specified by:
      getAdmin in interface org.apache.samza.system.SystemFactory

    • getConsumer

      public org.apache.samza.system.SystemConsumer getConsumer(String systemName, org.apache.samza.config.Config config, org.apache.samza.metrics.MetricsRegistry registry)
      Specified by:
      getConsumer in interface org.apache.samza.system.SystemFactory

    • getControllerDiscoveryUrl

      public Optional<String> getControllerDiscoveryUrl(org.apache.samza.config.Config config)

    • createSystemProducer

      @Deprecated protected org.apache.samza.system.SystemProducer createSystemProducer(String primaryControllerColoD2ZKHost, String primaryControllerD2Service, String storeName, Version.PushType venicePushType, String samzaJobId, String runningFabric, org.apache.samza.config.Config config, Optional<SSLFactory> sslFactory, Optional<String> partitioners)
      Deprecated.
      Left in to maintain backward compatibility

    • createSystemProducer

      @Deprecated protected org.apache.samza.system.SystemProducer createSystemProducer(String primaryControllerColoD2ZKHost, String primaryControllerD2Service, String storeName, Version.PushType venicePushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, org.apache.samza.config.Config config, Optional<SSLFactory> sslFactory, Optional<String> partitioners)
      Deprecated.
      Left in to maintain backward compatibility

    • createSystemProducer

      protected VeniceSystemProducer createSystemProducer(String veniceChildD2ZkHost, String primaryControllerColoD2ZKHost, String primaryControllerD2Service, String storeName, Version.PushType venicePushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, org.apache.samza.config.Config config, Optional<SSLFactory> sslFactory, Optional<String> partitioners)
      Construct a new instance of VeniceSystemProducer
      Parameters:
      veniceChildD2ZkHost - D2 Zk Address where the components in the child colo are announcing themselves
      primaryControllerColoD2ZKHost - D2 Zk Address of the colo where the primary controller resides
      primaryControllerD2ServiceName - The service name that the primary controller uses to announce itself to D2
      storeName - The store to write to
      pushType - The PushType to use to write to the store
      samzaJobId - A unique id used to identify jobs that can concurrently write to the same store
      runningFabric - The colo where the job is running. It is used to find the best destination for the data to be written to
      verifyLatestProtocolPresent - Config to check whether the protocol versions used at runtime are valid in Venice backend
      factory - The VeniceSystemFactory object that was used to create this object
      config - A Config object that may be used by the factory implementation to create an overridden SystemProducer instance
      sslFactory - An optional SSLFactory that is used to communicate with other components using SSL
      partitioners - A list of comma-separated partitioners class names that are supported.

    • createSystemProducer

      protected VeniceSystemProducer createSystemProducer(String veniceChildD2ZkHost, String primaryControllerColoD2ZKHost, String primaryControllerD2Service, String storeName, Version.PushType venicePushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, Optional<SSLFactory> sslFactory, Optional<String> partitioners)
      Construct a new instance of VeniceSystemProducer
      Parameters:
      veniceChildD2ZkHost - D2 Zk Address where the components in the child colo are announcing themselves
      primaryControllerColoD2ZKHost - D2 Zk Address of the colo where the primary controller resides
      primaryControllerD2ServiceName - The service name that the primary controller uses to announce itself to D2
      storeName - The store to write to
      pushType - The PushType to use to write to the store
      samzaJobId - A unique id used to identify jobs that can concurrently write to the same store
      runningFabric - The colo where the job is running. It is used to find the best destination for the data to be written to
      verifyLatestProtocolPresent - Config to check whether the protocol versions used at runtime are valid in Venice backend
      factory - The VeniceSystemFactory object that was used to create this object
      sslFactory - An optional SSLFactory that is used to communicate with other components using SSL
      partitioners - A list of comma-separated partitioners class names that are supported.

    • getProducer

      public org.apache.samza.system.SystemProducer getProducer(String systemName, String storeName, boolean veniceAggregate, String pushTypeString, org.apache.samza.config.Config config)
      Samza table writer would directly call this function to create venice system producer instead of calling the general getProducer(String, Config, MetricsRegistry) function.

    • getProducer

      public org.apache.samza.system.SystemProducer getProducer(String systemName, org.apache.samza.config.Config config, org.apache.samza.metrics.MetricsRegistry registry)
      The core function of a SystemFactory; most Samza users would specify VeniceSystemFactory in the job config and Samza would invoke SystemFactory.getProducer(String, Config, MetricsRegistry) to create producers.
      Specified by:
      getProducer in interface org.apache.samza.system.SystemFactory

    • getClosableProducer

      public VeniceSystemProducer getClosableProducer(String systemName, org.apache.samza.config.Config config, org.apache.samza.metrics.MetricsRegistry registry)
      Convenience method to hide the ugliness of casting in just one place. Ideally, we would change the return type of getProducer(String, Config, MetricsRegistry) to VeniceSystemProducer but since there are existing users of this API, we are being extra careful not to disturb it. TODO: clean this up when we have the bandwidth to coordinate the refactoring with the existing users.

    • getNumberOfActiveSystemProducers

      public int getNumberOfActiveSystemProducers()
      Get the total number of active SystemProducer. The SystemProducer for push type: STREAM and BATCH will always be at active state; so if there is any real-time SystemProducer in the Samza task, the task will not be stopped even though all the stream reprocessing jobs have completed. Besides, a Samza task can not have a mix of BATCH push type and STREAM_REPROCESSING push type; otherwise, the Samza task can not be automatically stopped.

    • getOverallExecutionStatus

      public boolean getOverallExecutionStatus()
      Check whether all the stream reprocessing jobs have succeeded; return false if any of them fail.

    • endStreamReprocessingSystemProducer

      public void endStreamReprocessingSystemProducer(org.apache.samza.system.SystemProducer systemProducer, boolean jobSucceed)
      RouterBasedPushMonitor will update the status of a SystemProducer with push type STREAM_REPROCESSING: END_OF_PUSH_RECEIVED: isActive -> false; isStreamReprocessingJobSucceeded -> true COMPLETED: isActive -> false; isStreamReprocessingJobSucceeded -> true ERROR: isActive -> false; isStreamReprocessingJobSucceeded -> false For all the other push job status, SystemProducer status will not be updated.