org.apache.beam.sdk.io.solace

Class SolaceIO.Read<T>

java.lang.Object

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

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

All Implemented Interfaces:

java.io.Serializable, HasDisplayData

Enclosing class:

SolaceIO

public static class SolaceIO.Read<T>
extends PTransform<PBegin,PCollection<T>>

See Also:

Serialized Form

Field Summary

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

annotations, displayData, name, resourceHints

Method Summary

Modifier and Type	Method and Description
PCollection<T>	expand(PBegin input) Override this method to specify how this PTransform should be expanded on the given InputT.
SolaceIO.Read<T>	from(Solace.Queue queue) Set the queue name to read from.
SolaceIO.Read<T>	from(Solace.Topic topic) Set the topic name to read from.
void	validate(@Nullable PipelineOptions options) Called before running the Pipeline to verify this transform is fully and correctly specified.
SolaceIO.Read<T>	withDeduplicateRecords(boolean deduplicateRecords) Optional, default: false.
SolaceIO.Read<T>	withMaxNumConnections(java.lang.Integer maxNumConnections) Optional.
SolaceIO.Read<T>	withSempClientFactory(SempClientFactory sempClientFactory) Set a factory that creates a SempClientFactory.
SolaceIO.Read<T>	withSessionServiceFactory(SessionServiceFactory sessionServiceFactory) Set a factory that creates a SessionService.
SolaceIO.Read<T>	withTimestampFn(SerializableFunction<T,Instant> timestampFn) The timestamp function, used for estimating the watermark, mapping the record T to an Instant
SolaceIO.Read<T>	withWatermarkIdleDurationThreshold(Duration idleDurationThreshold) Optional.

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

Methods inherited from class java.lang.Object

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

Method Detail

from

public SolaceIO.Read<T> from(Solace.Queue queue)

Set the queue name to read from. Use this or the `from(Topic)` method.

from

public SolaceIO.Read<T> from(Solace.Topic topic)

Set the topic name to read from. Use this or the `from(Queue)` method.

withTimestampFn

public SolaceIO.Read<T> withTimestampFn(SerializableFunction<T,Instant> timestampFn)

The timestamp function, used for estimating the watermark, mapping the record T to an Instant

Optional when using the no-arg SolaceIO.read() method. Defaults to SolaceIO.SENDER_TIMESTAMP_FUNCTION. When using the SolaceIO.read(TypeDescriptor, SerializableFunction, SerializableFunction) method, the function mapping from T to Instant has to be passed as an argument.

withMaxNumConnections

public SolaceIO.Read<T> withMaxNumConnections(java.lang.Integer maxNumConnections)

Optional. Sets the maximum number of connections to the broker. The actual number of sessions is determined by this and the number set by the runner. If not set, the number of sessions is determined by the runner. The number of connections created follows this logic: `numberOfConnections = min(maxNumConnections, desiredNumberOfSplits)`, where the `desiredNumberOfSplits` is set by the runner.

withWatermarkIdleDurationThreshold

public SolaceIO.Read<T> withWatermarkIdleDurationThreshold(Duration idleDurationThreshold)

Optional. Denotes the duration for which the watermark can be idle. If there are no incoming messages for this ‘idle’ period of time, the watermark is set to a timestamp representing a time earlier than now by the ‘idle’ period of time (e.g. if the ‘idle’ period of time is set to 30 seconds, and there is no new data incoming for 30 seconds, the watermark will be set to max(currentWatermark, now() - 30 seconds). The default watermark idle duration threshold is SolaceIO.DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD.

withDeduplicateRecords

public SolaceIO.Read<T> withDeduplicateRecords(boolean deduplicateRecords)

Optional, default: false. Set to deduplicate messages based on the XMLMessage.getApplicationMessageId() of the incoming BytesXMLMessage. If the field is null, then the XMLMessage.getReplicationGroupMessageId() will be used, which is always set by Solace.

withSempClientFactory

public SolaceIO.Read<T> withSempClientFactory(SempClientFactory sempClientFactory)

Set a factory that creates a SempClientFactory.

The factory `create()` method is invoked in each instance of an UnboundedSolaceReader. Created SempClient has to communicate with broker management API. It must support operations such as:

• query for outstanding backlog bytes in a Queue,
• query for metadata such as access-type of a Queue,
• requesting creation of new Queues.

An existing implementation of the SempClientFactory includes BasicAuthSempClientFactory which implements connection to the SEMP with the Basic Authentication method.

To use it, specify the credentials with the builder methods.

The format of the host is `[Protocol://]Host[:Port]`

 .withSempClientFactory(
         BasicAuthSempClientFactory.builder()
               .host("your-host-name-with-protocol") // e.g. "http://12.34.56.78:8080"
               .username("username")
               .password("password")
               .vpnName("vpn-name")
               .build())
 

withSessionServiceFactory

public SolaceIO.Read<T> withSessionServiceFactory(SessionServiceFactory sessionServiceFactory)

Set a factory that creates a SessionService.

The factory `create()` method is invoked in each instance of an UnboundedSolaceReader. Created SessionService has to be able to:

• initialize a connection with the broker,
• check liveliness of the connection,
• close the connection,
• create a MessageReceiver.

An existing implementation of the SempClientFactory includes BasicAuthJcsmpSessionService which implements the Basic Authentication to Solace. *

To use it, specify the credentials with the builder methods. *

The host is the IPv4 or IPv6 or host name of the appliance. IPv6 addresses must be encoded in brackets ([]). For example, "12.34.56.78", or "[fe80::1]". If connecting to a non-default port, it can be specified here using the "Host:Port" format. For example, "12.34.56.78:4444", or "[fe80::1]:4444".

 BasicAuthJcsmpSessionServiceFactory.builder()
     .host("your-host-name")
           // e.g. "12.34.56.78", or "[fe80::1]", or "12.34.56.78:4444"
     .username("semp-username")
     .password("semp-password")
     .vpnName("vpn-name")
     .build()));
 

validate

public void validate(@Nullable PipelineOptions options)

Description copied from class: PTransform

Called before running the Pipeline to verify this transform is fully and correctly specified.

By default, does nothing.

Overrides:

validate in class PTransform<PBegin,PCollection<T>>

expand

public PCollection<T> expand(PBegin 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<PBegin,PCollection<T>>