code: @Override
  	public void connect() {
  		try {
  			disconnect();
  
  			// Create topic admin
  			TopicAdminSettings.Builder topicBuilder = TopicAdminSettings.newBuilder();
  			if (credentialsProvider != null) {
  				topicBuilder.setCredentialsProvider(credentialsProvider);
  			}
  			if (executorProvider != null) {
  				topicBuilder.setExecutorProvider(executorProvider);
  			} else {
  				topicBuilder.setExecutorProvider(defaultExecutorProvider);
  			}
  			if (headerProvider != null) {
  				topicBuilder.setHeaderProvider(headerProvider);
  			}
  			if (transportChannelProvider != null) {
  				topicBuilder.setTransportChannelProvider(transportChannelProvider);
  			}
  			if (clock != null) {
  				topicBuilder.setClock(clock);
  			}
  			topicAdmin = TopicAdminClient.create(topicBuilder.build());
  
  			// Create subscription admin
  			SubscriptionAdminSettings.Builder subscriptionBuilder = SubscriptionAdminSettings.newBuilder();
  			if (credentialsProvider != null) {
  				subscriptionBuilder.setCredentialsProvider(credentialsProvider);
  			}
  			if (executorProvider != null) {
  				subscriptionBuilder.setExecutorProvider(executorProvider);
  			} else {
  				subscriptionBuilder.setExecutorProvider(defaultExecutorProvider);
  			}
  			if (headerProvider != null) {
  				subscriptionBuilder.setHeaderProvider(headerProvider);
  			}
  			if (transportChannelProvider != null) {
  				subscriptionBuilder.setTransportChannelProvider(transportChannelProvider);
  			}
  			if (clock != null) {
  				subscriptionBuilder.setClock(clock);
  			}
  			subscriptionAdmin = SubscriptionAdminClient.create(subscriptionBuilder.build());
  
  			connected();
  
  			connected.set(true);
  		} catch (Exception cause) {
  			reconnect(cause);
  		}
  	}
docs: --- CONNECT ---