org.apache.beam.sdk.io.solace

Class SolaceIO.Write<T>

java.lang.Object

org.apache.beam.sdk.transforms.PTransform<PCollection<T>,SolaceOutput>

org.apache.beam.sdk.io.solace.SolaceIO.Write<T>

All Implemented Interfaces:

java.io.Serializable, HasDisplayData

Enclosing class:

SolaceIO

public abstract static class SolaceIO.Write<T>
extends PTransform<PCollection<T>,SolaceOutput>

See Also:

Serialized Form

Field Summary

Fields

Modifier and Type	Field and Description
static TupleTag<BadRecord>	FAILED_PUBLISH_TAG
static TupleTag<Solace.PublishResult>	SUCCESSFUL_PUBLISH_TAG

Fields inherited from class org.apache.beam.sdk.transforms.PTransform

annotations, displayData, name, resourceHints

Constructor Summary

Constructors

Constructor and Description
Write()

Method Summary

Modifier and Type	Method and Description
SolaceOutput	expand(PCollection<T> input) Override this method to specify how this PTransform should be expanded on the given InputT.
SolaceIO.Write<T>	publishLatencyMetrics() Publish latency metrics using Beam metrics.
SolaceIO.Write<T>	to(Solace.Queue queue) Write to a Solace queue.
SolaceIO.Write<T>	to(Solace.Topic topic) Write to a Solace topic.
SolaceIO.Write<T>	withDeliveryMode(DeliveryMode deliveryMode) Set the delivery mode.
SolaceIO.Write<T>	withErrorHandler(ErrorHandler<BadRecord,?> errorHandler) An optional error handler for handling records that failed to publish to Solace.
SolaceIO.Write<T>	withNumberOfClientsPerWorker(int numberOfClientsPerWorker) The number of clients that each worker will create.
SolaceIO.Write<T>	withNumShards(int numShards) The number of workers used by the job to write to Solace.
SolaceIO.Write<T>	withSessionServiceFactory(SessionServiceFactory factory) Set the provider used to obtain the properties to initialize a new session in the broker.
SolaceIO.Write<T>	withSubmissionMode(SolaceIO.SubmissionMode submissionMode) This setting controls the JCSMP property MESSAGE_CALLBACK_ON_REACTOR.
SolaceIO.Write<T>	withWriterType(SolaceIO.WriterType writerType) Set the type of writer used by the connector.

Methods inherited from class org.apache.beam.sdk.transforms.PTransform

addAnnotation, compose, compose, getAdditionalInputs, getAnnotations, getDefaultOutputCoder, getDefaultOutputCoder, getDefaultOutputCoder, getKindString, getName, getResourceHints, populateDisplayData, setDisplayData, setResourceHints, toString, validate, validate

Methods inherited from class java.lang.Object

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

Field Detail

FAILED_PUBLISH_TAG

public static final TupleTag<BadRecord> FAILED_PUBLISH_TAG

SUCCESSFUL_PUBLISH_TAG

public static final TupleTag<Solace.PublishResult> SUCCESSFUL_PUBLISH_TAG

Constructor Detail

Write

public Write()

Method Detail

to

public SolaceIO.Write<T> to(Solace.Topic topic)

Write to a Solace topic.

The topic does not need to exist before launching the pipeline.

This will write all records to the same topic, ignoring their destination field.

Optional. If not specified, the connector will use dynamic destinations based on the destination field of Solace.Record.

to

public SolaceIO.Write<T> to(Solace.Queue queue)

Write to a Solace queue.

The queue must exist prior to launching the pipeline.

This will write all records to the same queue, ignoring their destination field.

Optional. If not specified, the connector will use dynamic destinations based on the destination field of Solace.Record.

withNumShards

public SolaceIO.Write<T> withNumShards(int numShards)

The number of workers used by the job to write to Solace.

This is optional, the default value is 20.

This is the maximum value that the job would use, but depending on the amount of data, the actual number of writers may be lower than this value. With the Dataflow runner, the connector will as maximum this number of VMs in the job (but the job itself may use more VMs).

Set this number taking into account the limits in the number of clients in your Solace cluster, and the need for performance when writing to Solace (more workers will achieve higher throughput).

withNumberOfClientsPerWorker

public SolaceIO.Write<T> withNumberOfClientsPerWorker(int numberOfClientsPerWorker)

The number of clients that each worker will create.

This is optional, the default number is 4.

The number of clients is per worker. If there are more than one worker, the number of clients will be multiplied by the number of workers. With the Dataflow runner, this will be the number of clients created per VM. The clients will be re-used across different threads in the same worker.

Set this number in combination with withNumShards(int), to ensure that the limit for number of clients in your Solace cluster is not exceeded.

Normally, using a higher number of clients with fewer workers will achieve better throughput at a lower cost, since the workers are better utilized. A good rule of thumb to use is setting as many clients per worker as vCPUs the worker has.

withDeliveryMode

public SolaceIO.Write<T> withDeliveryMode(DeliveryMode deliveryMode)

Set the delivery mode. This is optional, the default value is DIRECT.

For more details, see https://docs.solace.com/API/API-Developer-Guide/Message-Delivery-Modes.htm

publishLatencyMetrics

public SolaceIO.Write<T> publishLatencyMetrics()

Publish latency metrics using Beam metrics.

Latency metrics are only available if withDeliveryMode(DeliveryMode) is set to PERSISTENT. In that mode, latency is measured for each single message, as the time difference between the message creation and the reception of the publishing confirmation.

For the batched writer, the creation time is set for every message in a batch shortly before the batch is submitted. So the latency is very close to the actual publishing latency, and it does not take into account the time spent waiting for the batch to be submitted.

This is optional, the default value is false (don't publish latency metrics).

withSubmissionMode

public SolaceIO.Write<T> withSubmissionMode(SolaceIO.SubmissionMode submissionMode)

This setting controls the JCSMP property MESSAGE_CALLBACK_ON_REACTOR. Optional.

For full details, please check https://docs.solace.com/API/API-Developer-Guide/Java-API-Best-Practices.htm.

The Solace JCSMP client libraries can dispatch messages using three different modes:

One of the modes dispatches messages directly from the same thread that is doing the rest of I/O work. This mode favors lower latency but lower throughput. Set this to LOWER_LATENCY to use that mode (MESSAGE_CALLBACK_ON_REACTOR set to True).

Another mode uses a parallel thread to accumulate and dispatch messages. This mode favors higher throughput but also has higher latency. Set this to HIGHER_THROUGHPUT to use that mode. This is the default mode (MESSAGE_CALLBACK_ON_REACTOR set to False).

If you prefer to have full control over all the JCSMP properties, set this to CUSTOM, and override the classes SessionServiceFactory and SessionService to have full control on how to create the JCSMP sessions and producers used by the connector.

This is optional, the default value is HIGHER_THROUGHPUT.

withWriterType

public SolaceIO.Write<T> withWriterType(SolaceIO.WriterType writerType)

Set the type of writer used by the connector. Optional.

The Solace writer can either use the JCSMP modes in streaming or batched.

In streaming mode, the publishing latency will be lower, but the throughput will also be lower.

With the batched mode, messages are accumulated until a batch size of 50 is reached, or UnboundedBatchedSolaceWriter.ACKS_FLUSHING_INTERVAL_SECS seconds have elapsed since the first message in the batch was received. The 50 messages are sent to Solace in a single batch. This writer offers higher throughput but higher publishing latency, as messages can be held up for up to UnboundedBatchedSolaceWriter.ACKS_FLUSHING_INTERVAL_SECS5seconds until they are published.

Notice that this is the message publishing latency, not the end-to-end latency. For very large scale pipelines, you will probably prefer to use the HIGHER_THROUGHPUT mode, as with lower throughput messages will accumulate in the pipeline, and the end-to-end latency may actually be higher.

This is optional, the default is the BATCHED writer.

withSessionServiceFactory

public SolaceIO.Write<T> withSessionServiceFactory(SessionServiceFactory factory)

Set the provider used to obtain the properties to initialize a new session in the broker.

This provider should define the destination host where the broker is listening, and all the properties related to authentication (base auth, client certificate, etc.).

withErrorHandler

public SolaceIO.Write<T> withErrorHandler(ErrorHandler<BadRecord,?> errorHandler)

An optional error handler for handling records that failed to publish to Solace.

If provided, this error handler will be invoked for each record that could not be successfully published. The error handler can implement custom logic for dealing with failed records, such as writing them to a dead-letter queue or logging them.

If no error handler is provided, failed records will be ignored.

expand

public SolaceOutput expand(PCollection<T> input)

Description copied from class: PTransform

Override this method to specify how this PTransform should be expanded on the given InputT.

NOTE: This method should not be called directly. Instead apply the PTransform should be applied to the InputT using the apply method.

Composite transforms, which are defined in terms of other transforms, should return the output of one of the composed transforms. Non-composite transforms, which do not apply any transforms internally, should return a new unbound output and register evaluators (via backend-specific registration methods).

Specified by:

expand in class PTransform<PCollection<T>,SolaceOutput>