ClientBuilder.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.client.api;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.time.Clock;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;

/**
 * Builder interface that is used to configure and construct a {@link PulsarClient} instance.
 *
 * @since 2.0.0
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public interface ClientBuilder extends Serializable, Cloneable {

    /**
     * Construct the final {@link PulsarClient} instance.
     *
     * @return the new {@link PulsarClient} instance
     */
    PulsarClient build() throws PulsarClientException;

    /**
     * Load the configuration from provided <tt>config</tt> map.
     *
     * <p>Example:
     *
     * <pre>
     * {@code
     * Map<String, Object> config = new HashMap<>();
     * config.put("serviceUrl", "pulsar://localhost:6650");
     * config.put("numIoThreads", 20);
     *
     * ClientBuilder builder = ...;
     * builder = builder.loadConf(config);
     *
     * PulsarClient client = builder.build();
     * }
     * </pre>
     *
     * @param config
     *            configuration to load
     * @return the client builder instance
     */
    ClientBuilder loadConf(Map<String, Object> config);

    /**
     * Create a copy of the current client builder.
     *
     * <p>Cloning the builder can be used to share an incomplete configuration and specialize it multiple times. For
     * example:
     *
     * <pre>{@code
     * ClientBuilder builder = PulsarClient.builder()
     *               .ioThreads(8)
     *               .listenerThreads(4);
     *
     * PulsarClient client1 = builder.clone()
     *                  .serviceUrl("pulsar://localhost:6650").build();
     * PulsarClient client2 = builder.clone()
     *                  .serviceUrl("pulsar://other-host:6650").build();
     * }</pre>
     *
     * @return a clone of the client builder instance
     */
    ClientBuilder clone();

    /**
     * Configure the service URL for the Pulsar service.
     *
     * <p>This parameter is required.
     *
     * <p>Examples:
     * <ul>
     * <li>{@code pulsar://my-broker:6650} for regular endpoint</li>
     * <li>{@code pulsar+ssl://my-broker:6651} for TLS encrypted endpoint</li>
     * </ul>
     *
     * @param serviceUrl
     *            the URL of the Pulsar service that the client should connect to
     * @return the client builder instance
     */
    ClientBuilder serviceUrl(String serviceUrl);

    /**
     * Configure the service URL provider for Pulsar service.
     *
     * <p>Instead of specifying a static service URL string (with {@link #serviceUrl(String)}), an application
     * can pass a {@link ServiceUrlProvider} instance that dynamically provide a service URL.
     *
     * @param serviceUrlProvider
     *            the provider instance
     * @return the client builder instance
     */
    ClientBuilder serviceUrlProvider(ServiceUrlProvider serviceUrlProvider);

    /**
     * Configure the listenerName that the broker will return the corresponding `advertisedListener`.
     *
     * @param name the listener name
     * @return the client builder instance
     */
    ClientBuilder listenerName(String name);

    /**
     * Set the authentication provider to use in the Pulsar client instance.
     *
     * <p>Example:
     * <pre>{@code
     * PulsarClient client = PulsarClient.builder()
     *         .serviceUrl("pulsar+ssl://broker.example.com:6651/")
     *         .authentication(
     *               AuthenticationFactory.TLS("/my/cert/file", "/my/key/file")
     *         .build();
     * }</pre>
     *
     * <p>For token based authentication, this will look like:
     * <pre>{@code
     * AuthenticationFactory
     *      .token("eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJKb2UifQ.ipevRNuRP6HflG8cFKnmUPtypruRC4fb1DWtoLL62SY")
     * }</pre>
     *
     * @param authentication
     *            an instance of the {@link Authentication} provider already constructed
     * @return the client builder instance
     */
    ClientBuilder authentication(Authentication authentication);

    /**
     * Configure the authentication provider to use in the Pulsar client instance.
     *
     * <p>Example:
     * <pre>
     * <code>
     * PulsarClient client = PulsarClient.builder()
     *          .serviceUrl("pulsar+ssl://broker.example.com:6651/)
     *          .authentication(
     *              "org.apache.pulsar.client.impl.auth.AuthenticationTls",
     *              "tlsCertFile:/my/cert/file,tlsKeyFile:/my/key/file")
     *          .build();
     * </code>
     * </pre>
     *
     * @param authPluginClassName
     *            name of the Authentication-Plugin you want to use
     * @param authParamsString
     *            string which represents parameters for the Authentication-Plugin, e.g., "key1:val1,key2:val2"
     * @return the client builder instance
     * @throws UnsupportedAuthenticationException
     *             failed to instantiate specified Authentication-Plugin
     */
    ClientBuilder authentication(String authPluginClassName, String authParamsString)
            throws UnsupportedAuthenticationException;

    /**
     * Configure the authentication provider to use in the Pulsar client instance
     * using a config map.
     *
     * <p>Example:
     * <pre>{@code
     * Map<String, String> conf = new TreeMap<>();
     * conf.put("tlsCertFile", "/my/cert/file");
     * conf.put("tlsKeyFile", "/my/key/file");
     *
     * PulsarClient client = PulsarClient.builder()
     *          .serviceUrl("pulsar+ssl://broker.example.com:6651/)
     *          .authentication(
     *              "org.apache.pulsar.client.impl.auth.AuthenticationTls", conf)
     *          .build();
     * }</pre>
     *
     * @param authPluginClassName
     *            name of the Authentication-Plugin you want to use
     * @param authParams
     *            map which represents parameters for the Authentication-Plugin
     * @return the client builder instance
     * @throws UnsupportedAuthenticationException
     *             failed to instantiate specified Authentication-Plugin
     */
    ClientBuilder authentication(String authPluginClassName, Map<String, String> authParams)
            throws UnsupportedAuthenticationException;

    /**
     * Set the operation timeout <i>(default: 30 seconds)</i>.
     *
     * <p>Producer-create, subscribe and unsubscribe operations will be retried until this interval, after which the
     * operation will be marked as failed
     *
     * @param operationTimeout
     *            operation timeout
     * @param unit
     *            time unit for {@code operationTimeout}
     * @return the client builder instance
     */
    ClientBuilder operationTimeout(int operationTimeout, TimeUnit unit);

    /**
     * Set lookup timeout <i>(default: matches operation timeout)</i>
     *
     * Lookup operations have a different load pattern to other operations. They can be handled by any broker, are not
     * proportional to throughput, and are harmless to retry. Given this, it makes sense to allow them to retry longer
     * than normal operation, especially if they experience a timeout.
     *
     * By default this is set to match operation timeout. This is to maintain legacy behaviour. However, in practice
     * it should be set to 5-10x the operation timeout.
     *
     * @param lookupTimeout
     *            lookup timeout
     * @param unit
     *            time unit for {@code lookupTimeout}
     * @return the client builder instance
     */
    ClientBuilder lookupTimeout(int lookupTimeout, TimeUnit unit);

    /**
     * Set the number of threads to be used for handling connections to brokers <i>(default: 1 thread)</i>.
     *
     * @param numIoThreads the number of IO threads
     * @return the client builder instance
     */
    ClientBuilder ioThreads(int numIoThreads);

    /**
     * Set the number of threads to be used for message listeners <i>(default: 1 thread)</i>.
     *
     * <p>The listener thread pool is shared across all the consumers and readers that are
     * using a "listener" model to get messages. For a given consumer, the listener will be
     * always invoked from the same thread, to ensure ordering.
     *
     * @param numListenerThreads the number of listener threads
     * @return the client builder instance
     */
    ClientBuilder listenerThreads(int numListenerThreads);

    /**
     * Sets the max number of connection that the client library will open to a single broker.
     *
     * <p>By default, the connection pool will use a single connection for all the producers and consumers.
     * Increasing this parameter may improve throughput when using many producers over a high latency connection.
     *
     * @param connectionsPerBroker
     *            max number of connections per broker (needs to be greater than 0)
     * @return the client builder instance
     */
    ClientBuilder connectionsPerBroker(int connectionsPerBroker);

    /**
     * Configure whether to use TCP no-delay flag on the connection, to disable Nagle algorithm.
     *
     * <p>No-delay features make sure packets are sent out on the network as soon as possible, and it's critical
     * to achieve low latency publishes. On the other hand, sending out a huge number of small packets
     * might limit the overall throughput, so if latency is not a concern,
     * it's advisable to set the <code>useTcpNoDelay</code> flag to false.
     *
     * <p>Default value is true.
     *
     * @param enableTcpNoDelay whether to enable TCP no-delay feature
     * @return the client builder instance
     */
    ClientBuilder enableTcpNoDelay(boolean enableTcpNoDelay);

    /**
     * Configure whether to use TLS encryption on the connection
     * <i>(default: true if serviceUrl starts with "pulsar+ssl://", false otherwise)</i>.
     *
     * @param enableTls
     * @deprecated use "pulsar+ssl://" in serviceUrl to enable
     * @return the client builder instance
     */
    @Deprecated
    ClientBuilder enableTls(boolean enableTls);

    /**
     * Set the path to the trusted TLS certificate file.
     *
     * @param tlsTrustCertsFilePath
     * @return the client builder instance
     */
    ClientBuilder tlsTrustCertsFilePath(String tlsTrustCertsFilePath);

    /**
     * Configure whether the Pulsar client accept untrusted TLS certificate from broker <i>(default: false)</i>.
     *
     * @param allowTlsInsecureConnection whether to accept a untrusted TLS certificate
     * @return the client builder instance
     */
    ClientBuilder allowTlsInsecureConnection(boolean allowTlsInsecureConnection);

    /**
     * It allows to validate hostname verification when client connects to broker over tls. It validates incoming x509
     * certificate and matches provided hostname(CN/SAN) with expected broker's host name. It follows RFC 2818, 3.1.
     * Server Identity hostname verification.
     *
     * @see <a href="https://tools.ietf.org/html/rfc2818">RFC 818</a>
     *
     * @param enableTlsHostnameVerification whether to enable TLS hostname verification
     * @return the client builder instance
     */
    ClientBuilder enableTlsHostnameVerification(boolean enableTlsHostnameVerification);

    /**
     * If Tls is enabled, whether use KeyStore type as tls configuration parameter.
     * False means use default pem type configuration.
     *
     * @param useKeyStoreTls
     * @return the client builder instance
     */
    ClientBuilder useKeyStoreTls(boolean useKeyStoreTls);

    /**
     * The name of the security provider used for SSL connections.
     * Default value is the default security provider of the JVM.
     *
     * @param sslProvider
     * @return the client builder instance
     */
    ClientBuilder sslProvider(String sslProvider);

    /**
     * The file format of the trust store file.
     *
     * @param tlsTrustStoreType
     * @return the client builder instance
     */
    ClientBuilder tlsTrustStoreType(String tlsTrustStoreType);

    /**
     * The location of the trust store file.
     *
     * @param tlsTrustStorePath
     * @return the client builder instance
     */
    ClientBuilder tlsTrustStorePath(String tlsTrustStorePath);

    /**
     * The store password for the key store file.
     *
     * @param tlsTrustStorePassword
     * @return the client builder instance
     */
    ClientBuilder tlsTrustStorePassword(String tlsTrustStorePassword);

    /**
     * A list of cipher suites.
     * This is a named combination of authentication, encryption, MAC and key exchange algorithm
     * used to negotiate the security settings for a network connection using TLS or SSL network protocol.
     * By default all the available cipher suites are supported.
     *
     * @param tlsCiphers
     * @return the client builder instance
     */
    ClientBuilder tlsCiphers(Set<String> tlsCiphers);

    /**
     * The SSL protocol used to generate the SSLContext.
     * Default setting is TLS, which is fine for most cases.
     * Allowed values in recent JVMs are TLS, TLSv1.1 and TLSv1.2. SSL, SSLv2.
     *
     * @param tlsProtocols
     * @return the client builder instance
     */
    ClientBuilder tlsProtocols(Set<String> tlsProtocols);

    /**
     * Configure a limit on the amount of direct memory that will be allocated by this client instance.
     * <p>
     * <b>Note: at this moment this is only limiting the memory for producers.</b>
     * <p>
     * Setting this to 0 will disable the limit.
     *
     * @param memoryLimit
     *            the limit
     * @param unit
     *            the memory limit size unit
     * @return the client builder instance
     */
    ClientBuilder memoryLimit(long memoryLimit, SizeUnit unit);

    /**
     * Set the interval between each stat info <i>(default: 60 seconds)</i> Stats will be activated with positive
     * statsInterval It should be set to at least 1 second.
     *
     * @param statsInterval
     *            the interval between each stat info
     * @param unit
     *            time unit for {@code statsInterval}
     * @return the client builder instance
     */
    ClientBuilder statsInterval(long statsInterval, TimeUnit unit);

    /**
     * Number of concurrent lookup-requests allowed to send on each broker-connection to prevent overload on broker.
     * <i>(default: 5000)</i> It should be configured with higher value only in case of it requires to produce/subscribe
     * on thousands of topic using created {@link PulsarClient}.
     *
     * @param maxConcurrentLookupRequests
     * @return the client builder instance
     */
    ClientBuilder maxConcurrentLookupRequests(int maxConcurrentLookupRequests);

    /**
     * Number of max lookup-requests allowed on each broker-connection to prevent overload on broker.
     * <i>(default: 50000)</i> It should be bigger than maxConcurrentLookupRequests.
     * Requests that inside maxConcurrentLookupRequests already send to broker, and requests beyond
     * maxConcurrentLookupRequests and under maxLookupRequests will wait in each client cnx.
     *
     * @param maxLookupRequests
     * @return the client builder instance
     */
    ClientBuilder maxLookupRequests(int maxLookupRequests);

    /**
     * Set the maximum number of times a lookup-request to a broker will be redirected.
     *
     * @since 2.6.0
     * @param maxLookupRedirects the maximum number of redirects
     * @return the client builder instance
     */
    ClientBuilder maxLookupRedirects(int maxLookupRedirects);

    /**
     * Set max number of broker-rejected requests in a certain time-frame (30 seconds) after which current connection
     * will be closed and client creates a new connection that give chance to connect a different broker <i>(default:
     * 50)</i>.
     *
     * @param maxNumberOfRejectedRequestPerConnection
     * @return the client builder instance
     */
    ClientBuilder maxNumberOfRejectedRequestPerConnection(int maxNumberOfRejectedRequestPerConnection);

    /**
     * Set keep alive interval for each client-broker-connection. <i>(default: 30 seconds)</i>.
     *
     * @param keepAliveInterval
     * @param unit the time unit in which the keepAliveInterval is defined
     * @return the client builder instance
     */
    ClientBuilder keepAliveInterval(int keepAliveInterval, TimeUnit unit);

    /**
     * Set the duration of time to wait for a connection to a broker to be established. If the duration passes without a
     * response from the broker, the connection attempt is dropped.
     *
     * @since 2.3.0
     * @param duration
     *            the duration to wait
     * @param unit
     *            the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder connectionTimeout(int duration, TimeUnit unit);

    /**
     * Set the duration of time for a backoff interval.
     *
     * @param duration the duration of the interval
     * @param unit the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder startingBackoffInterval(long duration, TimeUnit unit);

    /**
     * Set the maximum duration of time for a backoff interval.
     *
     * @param duration the duration of the interval
     * @param unit the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder maxBackoffInterval(long duration, TimeUnit unit);

    /**
     * Option to enable busy-wait settings. Default is false.
     *
     * <b>WARNING</b>: This option will enable spin-waiting on executors and IO threads in order to reduce latency
     * during context switches. The spinning will consume 100% CPU even when the broker is not doing any work. It
     * is recommended to reduce the number of IO threads and BK client threads to only have few CPU cores busy.
     *
     * @param enableBusyWait whether to enable busy wait
     * @return the client builder instance
     */
    ClientBuilder enableBusyWait(boolean enableBusyWait);

    /**
     * The clock used by the pulsar client.
     *
     * <p>The clock is currently used by producer for setting publish timestamps.
     * {@link Clock#millis()} is called to retrieve current timestamp as the publish
     * timestamp when producers produce messages. The default clock is a system default zone
     * clock. So the publish timestamp is same as calling {@link System#currentTimeMillis()}.
     *
     * <p>Warning: the clock is used for TTL enforcement and timestamp based seeks.
     * so be aware of the impacts if you are going to use a different clock.
     *
     * @param clock the clock used by the pulsar client to retrieve time information
     * @return the client builder instance
     */
    ClientBuilder clock(Clock clock);

    /**
     * Proxy-service url when client would like to connect to broker via proxy. Client can choose type of proxy-routing
     * using {@link ProxyProtocol}.
     *
     * @param proxyServiceUrl proxy service url
     * @param proxyProtocol   protocol to decide type of proxy routing eg: SNI-routing
     * @return
     */
    ClientBuilder proxyServiceUrl(String proxyServiceUrl, ProxyProtocol proxyProtocol);

    /**
     * If enable transaction, start the transactionCoordinatorClient with pulsar client.
     *
     * @param enableTransaction whether enable transaction feature
     * @return
     */
    ClientBuilder enableTransaction(boolean enableTransaction);

    /**
     *  Set socks5 proxy address.
     * @param socks5ProxyAddress
     * @return
     */
    ClientBuilder socks5ProxyAddress(InetSocketAddress socks5ProxyAddress);

    /**
     *  Set socks5 proxy username.
     * @param socks5ProxyUsername
     * @return
     */
    ClientBuilder socks5ProxyUsername(String socks5ProxyUsername);

    /**
     *  Set socks5 proxy password.
     * @param socks5ProxyPassword
     * @return
     */
    ClientBuilder socks5ProxyPassword(String socks5ProxyPassword);
}