public SourceConfiguration()

in flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/SourceConfiguration.java [72:90]

• 18 lines of code
• 2 McCabe index (conditional complexity)

    public SourceConfiguration(Configuration configuration) {
        super(configuration);

        this.messageQueueCapacity = getInteger(ELEMENT_QUEUE_CAPACITY);
        this.partitionDiscoveryIntervalMs = get(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS);
        this.enableAutoAcknowledgeMessage = get(PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE);
        this.autoCommitCursorInterval = get(PULSAR_AUTO_COMMIT_CURSOR_INTERVAL);
        this.fetchOneMessageTime = getOptional(PULSAR_FETCH_ONE_MESSAGE_TIME).orElse(0);
        this.maxFetchTime = get(PULSAR_MAX_FETCH_TIME, Duration::ofMillis);
        this.maxFetchRecords = get(PULSAR_MAX_FETCH_RECORDS);
        this.verifyInitialOffsets = get(PULSAR_VERIFY_INITIAL_OFFSETS);
        this.subscriptionName = get(PULSAR_SUBSCRIPTION_NAME);
        this.subscriptionMode = get(PULSAR_SUBSCRIPTION_MODE);
        this.allowKeySharedOutOfOrderDelivery = get(PULSAR_ALLOW_KEY_SHARED_OUT_OF_ORDER_DELIVERY);
        this.enableSchemaEvolution = get(PULSAR_READ_SCHEMA_EVOLUTION);
        this.enableMetrics =
                get(PULSAR_ENABLE_SOURCE_METRICS) && get(PULSAR_STATS_INTERVAL_SECONDS) > 0;
        this.resetSubscriptionCursor = get(PULSAR_RESET_SUBSCRIPTION_CURSOR);
    }