Package com.linkedin.venice.samza

Class VeniceSystemProducer

java.lang.Object

com.linkedin.venice.samza.VeniceSystemProducer

All Implemented Interfaces:

Closeable, AutoCloseable, org.apache.samza.system.SystemProducer

public class VeniceSystemProducer
extends Object
implements org.apache.samza.system.SystemProducer, Closeable

VeniceSystemProducer defines the interfaces for Samza jobs to send data to Venice stores. 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. Version.PushType is Version.PushType.BATCH or Version.PushType.STREAM_REPROCESSING; 2. The child controller when either: a. The Venice system is deployed in a single-colo mode; or b. The Version.PushType is Version.PushType.STREAM and the job is configured to write data in NON_AGGREGATE mode

Constructor Summary

Constructors

Constructor	Description
VeniceSystemProducer(String discoveryUrl, String storeName, Version.PushType pushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, VeniceSystemFactory factory, Optional<SSLFactory> sslFactory, Optional<String> partitioners, Time time)
VeniceSystemProducer(String primaryControllerColoD2ZKHost, String primaryControllerD2ServiceName, String storeName, Version.PushType pushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, VeniceSystemFactory factory, Optional<SSLFactory> sslFactory, Optional<String> partitioners)	Deprecated.
VeniceSystemProducer(String primaryControllerColoD2ZKHost, String primaryControllerD2ServiceName, String storeName, Version.PushType pushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, VeniceSystemFactory factory, Optional<SSLFactory> sslFactory, Optional<String> partitioners, Time time)	Deprecated.
VeniceSystemProducer(String veniceChildD2ZkHost, String primaryControllerColoD2ZKHost, String primaryControllerD2ServiceName, String storeName, Version.PushType pushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, VeniceSystemFactory factory, Optional<SSLFactory> sslFactory, Optional<String> partitioners)	Construct a new instance of VeniceSystemProducer.
VeniceSystemProducer(String veniceChildD2ZkHost, String primaryControllerColoD2ZKHost, String primaryControllerD2ServiceName, String storeName, Version.PushType pushType, String samzaJobId, String runningFabric, boolean verifyLatestProtocolPresent, VeniceSystemFactory factory, Optional<SSLFactory> sslFactory, Optional<String> partitioners, Time time)	Construct a new instance of VeniceSystemProducer

Method Summary

Modifier and Type	Method	Description
void	applyAdditionalConfigs(Map<String,String> additionalConfigs)
void	close()
protected ControllerResponse	controllerRequestWithRetry(Supplier<ControllerResponse> supplier, int retryLimit)
protected Object	convertPartialUpdateToFullPut(Pair<Integer,Integer> schemaIds, Object incomingWriteValueObject)
CompletableFuture<Void>	delete(Object keyObject)
protected static void	extractConcurrentProducerConfig(Properties veniceWriterProperties, VeniceWriterOptions.Builder builder)
void	flush(String s)	Flushing the data to Venice store in case VeniceSystemProducer buffers message.
AbstractVeniceWriter<byte[],byte[],byte[]>	getInternalWriter()
String	getKafkaBootstrapServers()	Test methods
String	getRunningFabric()
String	getTopicName()
protected AbstractVeniceWriter<byte[],byte[],byte[]>	getVeniceWriter(VersionCreationResponse store)	This method is overridden and not used by LinkedIn internally.
protected AbstractVeniceWriter<byte[],byte[],byte[]>	getVeniceWriter(VersionCreationResponse store, Properties veniceWriterProperties)	Please don't remove this method since it is still being used by LinkedIn internally.
CompletableFuture<Void>	put(Object keyObject, Object valueObject)
void	register(String source)
protected CompletableFuture<Void>	send(Object keyObject, Object valueObject)
void	send(String source, org.apache.samza.system.OutgoingMessageEnvelope outgoingMessageEnvelope)
protected void	setControllerClient(D2ControllerClient controllerClient)
void	setExitMode(SamzaExitMode exitMode)
void	setRouterUrl(String routerUrl)
protected void	setupClientsAndReInitProvider()
void	start()
void	stop()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

VeniceSystemProducer

@Deprecated
public VeniceSystemProducer(String primaryControllerColoD2ZKHost,
 String primaryControllerD2ServiceName,
 String storeName,
 Version.PushType pushType,
 String samzaJobId,
 String runningFabric,
 boolean verifyLatestProtocolPresent,
 VeniceSystemFactory factory,
 Optional<SSLFactory> sslFactory,
 Optional<String> partitioners)

Deprecated.

VeniceSystemProducer

public VeniceSystemProducer(String veniceChildD2ZkHost,
 String primaryControllerColoD2ZKHost,
 String primaryControllerD2ServiceName,
 String storeName,
 Version.PushType pushType,
 String samzaJobId,
 String runningFabric,
 boolean verifyLatestProtocolPresent,
 VeniceSystemFactory factory,
 Optional<SSLFactory> sslFactory,
 Optional<String> partitioners)

Construct a new instance of VeniceSystemProducer. Equivalent to VeniceSystemProducer(java.lang.String,java.lang.String,java.lang.String,java.lang.String,com.linkedin.venice.meta.Version.PushType,java.lang.String,java.lang.String,boolean,com.linkedin.venice.samza.VeniceSystemFactory,java.util.Optional<com.linkedin.venice.security.SSLFactory>,java.util.Optional<java.lang.String>,com.linkedin.venice.utils.Time)

VeniceSystemProducer

@Deprecated
public VeniceSystemProducer(String primaryControllerColoD2ZKHost,
 String primaryControllerD2ServiceName,
 String storeName,
 Version.PushType pushType,
 String samzaJobId,
 String runningFabric,
 boolean verifyLatestProtocolPresent,
 VeniceSystemFactory factory,
 Optional<SSLFactory> sslFactory,
 Optional<String> partitioners,
 Time time)

Deprecated.

VeniceSystemProducer

public VeniceSystemProducer(String veniceChildD2ZkHost,
 String primaryControllerColoD2ZKHost,
 String primaryControllerD2ServiceName,
 String storeName,
 Version.PushType pushType,
 String samzaJobId,
 String runningFabric,
 boolean verifyLatestProtocolPresent,
 VeniceSystemFactory factory,
 Optional<SSLFactory> sslFactory,
 Optional<String> partitioners,
 Time time)

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 Version.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.

time - An object of type Time. It is helpful to be configurable for testing.

VeniceSystemProducer

public VeniceSystemProducer(String discoveryUrl,
 String storeName,
 Version.PushType pushType,
 String samzaJobId,
 String runningFabric,
 boolean verifyLatestProtocolPresent,
 VeniceSystemFactory factory,
 Optional<SSLFactory> sslFactory,
 Optional<String> partitioners,
 Time time)

Method Details

applyAdditionalConfigs

public void applyAdditionalConfigs(Map<String,String> additionalConfigs)

setRouterUrl

public void setRouterUrl(String routerUrl)

getRunningFabric

public String getRunningFabric()

controllerRequestWithRetry

protected ControllerResponse controllerRequestWithRetry(Supplier<ControllerResponse> supplier,
 int retryLimit)

getTopicName

public String getTopicName()

getVeniceWriter

protected AbstractVeniceWriter<byte[],byte[],byte[]> getVeniceWriter(VersionCreationResponse store)

This method is overridden and not used by LinkedIn internally. Please update the overridden method accordingly after modifying this method.

extractConcurrentProducerConfig

protected static void extractConcurrentProducerConfig(Properties veniceWriterProperties,
 VeniceWriterOptions.Builder builder)

getVeniceWriter

protected AbstractVeniceWriter<byte[],byte[],byte[]> getVeniceWriter(VersionCreationResponse store,
 Properties veniceWriterProperties)

Please don't remove this method since it is still being used by LinkedIn internally.

setupClientsAndReInitProvider

protected void setupClientsAndReInitProvider()

start

public void start()

Specified by:

start in interface org.apache.samza.system.SystemProducer

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

stop

public void stop()

Specified by:

stop in interface org.apache.samza.system.SystemProducer

register

public void register(String source)

Specified by:

register in interface org.apache.samza.system.SystemProducer

send

public void send(String source,
 org.apache.samza.system.OutgoingMessageEnvelope outgoingMessageEnvelope)

Specified by:

send in interface org.apache.samza.system.SystemProducer

send

protected CompletableFuture<Void> send(Object keyObject,
 Object valueObject)

put

public CompletableFuture<Void> put(Object keyObject,
 Object valueObject)

delete

public CompletableFuture<Void> delete(Object keyObject)

flush

public void flush(String s)

Flushing the data to Venice store in case VeniceSystemProducer buffers message.

Specified by:

flush in interface org.apache.samza.system.SystemProducer

Parameters:

s - String representing the source of the message. Currently, VeniceSystemProducer is not using this param.

convertPartialUpdateToFullPut

protected Object convertPartialUpdateToFullPut(Pair<Integer,Integer> schemaIds,
 Object incomingWriteValueObject)

setExitMode

public void setExitMode(SamzaExitMode exitMode)

getKafkaBootstrapServers

public String getKafkaBootstrapServers()

Test methods

getInternalWriter

public AbstractVeniceWriter<byte[],byte[],byte[]> getInternalWriter()

setControllerClient

protected void setControllerClient(D2ControllerClient controllerClient)