AHeise commented on a change in pull request #16900:
URL: https://github.com/apache/flink/pull/16900#discussion_r694562472
##########
File path:
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
##########
@@ -234,25 +239,87 @@ public static void checkConfigurations(Configuration
configuration) {
PulsarClient client, Schema<T> schema,
ConsumerConfigurationData<T> config)
throws PulsarClientException {
// ConsumerBuilder don't support using the given
ConsumerConfigurationData directly.
- ConsumerBuilder<T> consumerBuilder =
client.newConsumer(schema).loadConf(configMap(config));
+ ConsumerBuilder<T> builder = new
ConsumerBuilderImpl<>((PulsarClientImpl) client, schema);
- // Set some non-serializable fields.
- if (config.getMessageListener() != null) {
- consumerBuilder.messageListener(config.getMessageListener());
- }
- if (config.getConsumerEventListener() != null) {
-
consumerBuilder.consumerEventListener(config.getConsumerEventListener());
- }
- if (config.getCryptoKeyReader() != null) {
- consumerBuilder.cryptoKeyReader(config.getCryptoKeyReader());
- }
- if (config.getMessageCrypto() != null) {
- consumerBuilder.messageCrypto(config.getMessageCrypto());
- }
- if (config.getBatchReceivePolicy() != null) {
- consumerBuilder.batchReceivePolicy(config.getBatchReceivePolicy());
+ // Since pulsar don't expose the config constructor.
+ // We have to set the builder fields one by one.
+ setConfig(config.getTopicNames(), topics -> builder.topics(new
ArrayList<>(topics)));
+ setConfig(config.getTopicsPattern(), builder::topicsPattern);
+ setConfig(config.getSubscriptionName(), builder::subscriptionName);
+ setConfig(
+ config.getAckTimeoutMillis(),
+ millis -> builder.ackTimeout(millis, TimeUnit.MILLISECONDS));
+ setConfig(config.isAckReceiptEnabled(), builder::isAckReceiptEnabled);
+ setConfig(
+ config.getTickDurationMillis(),
+ millis -> builder.ackTimeoutTickTime(millis,
TimeUnit.MILLISECONDS));
+ setConfig(
+ config.getNegativeAckRedeliveryDelayMicros(),
+ micros -> builder.negativeAckRedeliveryDelay(micros,
TimeUnit.MICROSECONDS));
+ setConfig(config.getSubscriptionType(), builder::subscriptionType);
+ setConfig(config.getSubscriptionMode(), builder::subscriptionMode);
+ setConfig(config.getMessageListener(), builder::messageListener);
+ setConfig(config.getConsumerEventListener(),
builder::consumerEventListener);
+ setConfig(config.getCryptoKeyReader(), builder::cryptoKeyReader);
+ setConfig(config.getMessageCrypto(), builder::messageCrypto);
+ setConfig(config.getCryptoFailureAction(),
builder::cryptoFailureAction);
+ setConfig(config.getReceiverQueueSize(), builder::receiverQueueSize);
+ setConfig(
+ config.getAcknowledgementsGroupTimeMicros(),
+ micros -> builder.acknowledgmentGroupTime(micros,
TimeUnit.MICROSECONDS));
+ setConfig(config.getConsumerName(), builder::consumerName);
+ setConfig(config.getPriorityLevel(), builder::priorityLevel);
+ setConfig(config.getMaxPendingChunkedMessage(),
builder::maxPendingChunkedMessage);
+ setConfig(
+ config.isAutoAckOldestChunkedMessageOnQueueFull(),
+ builder::autoAckOldestChunkedMessageOnQueueFull);
+ setConfig(config.getProperties(), builder::properties);
+ setConfig(
+ config.getMaxTotalReceiverQueueSizeAcrossPartitions(),
+ builder::maxTotalReceiverQueueSizeAcrossPartitions);
+ setConfig(config.isReadCompacted(), builder::readCompacted);
+ setConfig(config.getPatternAutoDiscoveryPeriod(),
builder::patternAutoDiscoveryPeriod);
+ setConfig(config.getPatternAutoDiscoveryPeriod(),
builder::patternAutoDiscoveryPeriod);
+ setConfig(config.getSubscriptionInitialPosition(),
builder::subscriptionInitialPosition);
+ setConfig(config.getRegexSubscriptionMode(),
builder::subscriptionTopicsMode);
+ setConfig(config.isReplicateSubscriptionState(),
builder::replicateSubscriptionState);
+ setConfig(config.getDeadLetterPolicy(), builder::deadLetterPolicy);
+ setConfig(config.isAutoUpdatePartitions(),
builder::autoUpdatePartitions);
+ setConfig(
+ config.getAutoUpdatePartitionsIntervalSeconds(),
+ seconds ->
+ builder.autoUpdatePartitionsInterval(
+ Math.toIntExact(seconds), TimeUnit.SECONDS));
+ if (config.isResetIncludeHead()) {
+ builder.startMessageIdInclusive();
}
+ setConfig(config.getBatchReceivePolicy(), builder::batchReceivePolicy);
+ setConfig(config.getKeySharedPolicy(), builder::keySharedPolicy);
+ setConfig(config.isRetryEnable(), builder::enableRetry);
+ setConfig(config.isBatchIndexAckEnabled(),
builder::enableBatchIndexAcknowledgment);
+ setConfig(
+ config.getExpireTimeOfIncompleteChunkedMessageMillis(),
+ millis ->
+ builder.expireTimeOfIncompleteChunkedMessage(
+ millis, TimeUnit.MILLISECONDS));
+ setConfig(config.isPoolMessages(), builder::poolMessages);
- return consumerBuilder.subscribe();
+ return builder.subscribe();
+ }
+
+ private static <T> void setConfig(T value, java.util.function.Consumer<T>
consumer) {
+ if (value != null) {
+ if (value instanceof Collection) {
+ if (!((Collection<?>) value).isEmpty()) {
+ consumer.accept(value);
+ }
+ } else if (value instanceof Map) {
+ if (!((Map<?, ?>) value).isEmpty()) {
+ consumer.accept(value);
+ }
+ } else {
+ consumer.accept(value);
+ }
Review comment:
I would be fine with a separate commit in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
