You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ti...@apache.org on 2022/09/30 03:46:38 UTC
[flink] branch release-1.15 updated (dbfdebaa1c0 -> 33b290ec9c4)
This is an automated email from the ASF dual-hosted git repository.
tison pushed a change to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
from dbfdebaa1c0 [FLINK-28733][scripts] jobmanager.sh supports dynamic parameters
new 1ccf832d091 [FLINK-26182][Connector/Pulsar] Drop the embedded Pulsar runtime for in favor of mock Pulsar runtime with local metastore support.
new 2e4eb091fc7 [FLINK-26182][Connector/Pulsar] Enable the transaction in the Pulsar TestContainers in a better way.
new 022df77665e [FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
new 6075746cb3b [FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
new 33b290ec9c4 [FLINK-29381][Connector/Pulsar] Fixes the split assignment for Key Shared subscription.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../f4d91193-72ba-4ce4-ad83-98f780dce581 | 6 +
flink-connectors/flink-connector-pulsar/pom.xml | 2 -
...itAssigner.java => KeySharedSplitAssigner.java} | 65 +-
.../assigner/NonSharedSplitAssigner.java | 33 -
.../enumerator/assigner/SplitAssignerBase.java | 33 +
.../enumerator/assigner/SplitAssignerFactory.java | 4 +-
.../source/enumerator/topic/TopicPartition.java | 5 +
.../pulsar/source/enumerator/topic/TopicRange.java | 2 +-
.../connector/pulsar/sink/PulsarSinkITCase.java | 155 +--
.../pulsar/source/PulsarSourceITCase.java | 6 +-
.../pulsar/source/PulsarUnorderedSourceITCase.java | 62 +-
.../assigner/NonSharedSplitAssignerTest.java | 2 +-
.../pulsar/testutils/PulsarTestCommonUtils.java | 22 +-
.../pulsar/testutils/PulsarTestContext.java | 47 +-
.../pulsar/testutils/PulsarTestContextFactory.java | 5 +-
.../pulsar/testutils/PulsarTestEnvironment.java | 2 +-
.../pulsar/testutils/PulsarTestSuiteBase.java | 5 +-
.../cases/MultipleTopicTemplateContext.java | 134 ---
.../cases/SingleTopicConsumingContext.java | 132 ---
.../pulsar/testutils/runtime/PulsarRuntime.java | 17 +-
.../testutils/runtime/PulsarRuntimeOperator.java | 15 +-
.../runtime/container/PulsarContainerRuntime.java | 26 +-
.../runtime/embedded/PulsarEmbeddedRuntime.java | 184 ----
.../testutils/runtime/mock/PulsarMockRuntime.java | 5 +-
.../testutils/sink/PulsarPartitionDataReader.java | 93 ++
.../testutils/sink/PulsarSinkTestContext.java | 124 +++
.../testutils/sink/PulsarSinkTestSuiteBase.java | 30 +-
.../source}/KeyedPulsarPartitionDataWriter.java | 18 +-
.../{ => source}/PulsarPartitionDataWriter.java | 18 +-
.../testutils/source/PulsarSourceTestContext.java | 138 +++
.../source/UnorderedSourceTestSuiteBase.java} | 45 +-
.../source/cases/KeySharedSubscriptionContext.java | 80 ++
.../cases/MultipleTopicConsumingContext.java | 37 +-
.../cases/SharedSubscriptionContext.java} | 26 +-
.../source/cases/SingleTopicConsumingContext.java | 73 ++
.../test/resources/containers/txnStandalone.conf | 1042 --------------------
.../src/test/resources/docker/bootstrap.sh | 28 -
.../flink-end-to-end-tests-pulsar/pom.xml | 24 +-
...eOrderedE2ECase.java => PulsarSinkE2ECase.java} | 30 +-
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 5 +-
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 13 +-
.../pulsar/cases/KeySharedSubscriptionContext.java | 144 ---
.../pulsar/cases/SharedSubscriptionContext.java | 116 ---
.../FlinkContainerWithPulsarEnvironment.java | 8 +-
.../common/UnorderedSourceTestSuiteBase.java | 86 --
.../ExclusiveSubscriptionContext.java | 19 +-
.../FailoverSubscriptionContext.java | 19 +-
.../source/enumerator/NoOpEnumStateSerializer.java | 2 +-
48 files changed, 890 insertions(+), 2297 deletions(-)
copy flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/{NonSharedSplitAssigner.java => KeySharedSplitAssigner.java} (57%)
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicTemplateContext.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/embedded/PulsarEmbeddedRuntime.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarPartitionDataReader.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
copy flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/source/enumerator/NoOpEnumStateSerializer.java => flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestSuiteBase.java (50%)
rename {flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common => flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source}/KeyedPulsarPartitionDataWriter.java (77%)
rename flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/{ => source}/PulsarPartitionDataWriter.java (69%)
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
copy flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/{source/PulsarUnorderedSourceITCase.java => testutils/source/UnorderedSourceTestSuiteBase.java} (55%)
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
rename flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/{ => source}/cases/MultipleTopicConsumingContext.java (58%)
rename flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/{cases/SharedSubscriptionConsumingContext.java => source/cases/SharedSubscriptionContext.java} (60%)
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/resources/containers/txnStandalone.conf
delete mode 100755 flink-connectors/flink-connector-pulsar/src/test/resources/docker/bootstrap.sh
copy flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/{PulsarSourceOrderedE2ECase.java => PulsarSinkE2ECase.java} (65%)
delete mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
delete mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
delete mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
rename flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/{cases => source}/ExclusiveSubscriptionContext.java (71%)
rename flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/{cases => source}/FailoverSubscriptionContext.java (71%)
[flink] 02/05: [FLINK-26182][Connector/Pulsar] Enable the transaction in the Pulsar TestContainers in a better way.
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2e4eb091fc7ce0210195671840b9a01ce5f15027
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Sep 27 18:04:24 2022 +0800
[FLINK-26182][Connector/Pulsar] Enable the transaction in the Pulsar TestContainers in a better way.
---
.../runtime/container/PulsarContainerRuntime.java | 7 +-
.../test/resources/containers/txnStandalone.conf | 1042 --------------------
2 files changed, 4 insertions(+), 1045 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
index 10bd5120a46..910985338b9 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
@@ -90,15 +90,16 @@ public class PulsarContainerRuntime implements PulsarRuntime {
// Override the default configuration in container for enabling the Pulsar transaction.
container.withClasspathResourceMapping(
- "containers/txnStandalone.conf",
- "/pulsar/conf/standalone.conf",
- BindMode.READ_ONLY);
+ "docker/bootstrap.sh", "/pulsar/bin/bootstrap.sh", BindMode.READ_ONLY);
// Waiting for the Pulsar border is ready.
container.waitingFor(
forHttp("/admin/v2/namespaces/public/default")
.forPort(BROKER_HTTP_PORT)
.forStatusCode(200)
.withStartupTimeout(Duration.ofMinutes(5)));
+ // Set custom startup script.
+ container.withCommand("sh /pulsar/bin/bootstrap.sh");
+
// Start the Pulsar Container.
container.start();
// Append the output to this runtime logger. Used for local debug purpose.
diff --git a/flink-connectors/flink-connector-pulsar/src/test/resources/containers/txnStandalone.conf b/flink-connectors/flink-connector-pulsar/src/test/resources/containers/txnStandalone.conf
deleted file mode 100644
index 10437a71740..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/resources/containers/txnStandalone.conf
+++ /dev/null
@@ -1,1042 +0,0 @@
-#
-# 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.
-#
-
-### --- General broker settings --- ###
-
-# Zookeeper quorum connection string
-zookeeperServers=
-
-# Configuration Store connection string
-configurationStoreServers=
-
-brokerServicePort=6650
-
-# Port to use to server HTTP request
-webServicePort=8080
-
-# Hostname or IP address the service binds on, default is 0.0.0.0.
-bindAddress=0.0.0.0
-
-# Extra bind addresses for the service: <listener_name>:<scheme>://<host>:<port>,[...]
-bindAddresses=
-
-# Hostname or IP address the service advertises to the outside world. If not set, the value of InetAddress.getLocalHost().getHostName() is used.
-advertisedAddress=
-
-# Enable or disable the HAProxy protocol.
-haProxyProtocolEnabled=false
-
-# Number of threads to use for Netty IO. Default is set to 2 * Runtime.getRuntime().availableProcessors()
-numIOThreads=
-
-# Number of threads to use for ordered executor. The ordered executor is used to operate with zookeeper,
-# such as init zookeeper client, get namespace policies from zookeeper etc. It also used to split bundle. Default is 8
-numOrderedExecutorThreads=8
-
-# Number of threads to use for HTTP requests processing. Default is set to 2 * Runtime.getRuntime().availableProcessors()
-numHttpServerThreads=
-
-# Number of thread pool size to use for pulsar broker service.
-# The executor in thread pool will do basic broker operation like load/unload bundle, update managedLedgerConfig,
-# update topic/subscription/replicator message dispatch rate, do leader election etc.
-# Default is Runtime.getRuntime().availableProcessors()
-numExecutorThreadPoolSize=
-
-# Number of thread pool size to use for pulsar zookeeper callback service
-# The cache executor thread pool is used for restarting global zookeeper session.
-# Default is 10
-numCacheExecutorThreadPoolSize=10
-
-# Max concurrent web requests
-maxConcurrentHttpRequests=1024
-
-# Name of the cluster to which this broker belongs to
-clusterName=standalone
-
-# Enable cluster's failure-domain which can distribute brokers into logical region
-failureDomainsEnabled=false
-
-# Zookeeper session timeout in milliseconds
-zooKeeperSessionTimeoutMillis=30000
-
-# ZooKeeper operation timeout in seconds
-zooKeeperOperationTimeoutSeconds=30
-
-# ZooKeeper cache expiry time in seconds
-zooKeeperCacheExpirySeconds=300
-
-# Time to wait for broker graceful shutdown. After this time elapses, the process will be killed
-brokerShutdownTimeoutMs=60000
-
-# Flag to skip broker shutdown when broker handles Out of memory error
-skipBrokerShutdownOnOOM=false
-
-# Enable backlog quota check. Enforces action on topic when the quota is reached
-backlogQuotaCheckEnabled=true
-
-# How often to check for topics that have reached the quota
-backlogQuotaCheckIntervalInSeconds=60
-
-# Default per-topic backlog quota limit
-backlogQuotaDefaultLimitGB=10
-
-# Default per-topic backlog quota time limit in second, less than 0 means no limitation. default is -1.
-backlogQuotaDefaultLimitSecond=-1
-
-# Default ttl for namespaces if ttl is not already configured at namespace policies. (disable default-ttl with value 0)
-ttlDurationDefaultInSeconds=0
-
-# Enable the deletion of inactive topics. This parameter need to cooperate with the allowAutoTopicCreation parameter.
-# If brokerDeleteInactiveTopicsEnabled is set to true, we should ensure that allowAutoTopicCreation is also set to true.
-brokerDeleteInactiveTopicsEnabled=true
-
-# How often to check for inactive topics
-brokerDeleteInactiveTopicsFrequencySeconds=60
-
-# Allow you to delete a tenant forcefully.
-forceDeleteTenantAllowed=false
-
-# Allow you to delete a namespace forcefully.
-forceDeleteNamespaceAllowed=false
-
-# Max pending publish requests per connection to avoid keeping large number of pending
-# requests in memory. Default: 1000
-maxPendingPublishRequestsPerConnection=1000
-
-# How frequently to proactively check and purge expired messages
-messageExpiryCheckIntervalInMinutes=5
-
-# Check between intervals to see if max message size in topic policies has been updated.
-# Default is 60s
-maxMessageSizeCheckIntervalInSeconds=60
-
-# How long to delay rewinding cursor and dispatching messages when active consumer is changed
-activeConsumerFailoverDelayTimeMillis=1000
-
-# How long to delete inactive subscriptions from last consuming
-# When it is 0, inactive subscriptions are not deleted automatically
-subscriptionExpirationTimeMinutes=0
-
-# Enable subscription message redelivery tracker to send redelivery count to consumer (default is enabled)
-subscriptionRedeliveryTrackerEnabled=true
-
-# On KeyShared subscriptions, with default AUTO_SPLIT mode, use splitting ranges or
-# consistent hashing to reassign keys to new consumers
-subscriptionKeySharedUseConsistentHashing=false
-
-# On KeyShared subscriptions, number of points in the consistent-hashing ring.
-# The higher the number, the more equal the assignment of keys to consumers
-subscriptionKeySharedConsistentHashingReplicaPoints=100
-
-# How frequently to proactively check and purge expired subscription
-subscriptionExpiryCheckIntervalInMinutes=5
-
-# Set the default behavior for message deduplication in the broker
-# This can be overridden per-namespace. If enabled, broker will reject
-# messages that were already stored in the topic
-brokerDeduplicationEnabled=false
-
-# Maximum number of producer information that it's going to be
-# persisted for deduplication purposes
-brokerDeduplicationMaxNumberOfProducers=10000
-
-# Number of entries after which a dedup info snapshot is taken.
-# A bigger interval will lead to less snapshots being taken though it would
-# increase the topic recovery time, when the entries published after the
-# snapshot need to be replayed
-brokerDeduplicationEntriesInterval=1000
-
-# Time of inactivity after which the broker will discard the deduplication information
-# relative to a disconnected producer. Default is 6 hours.
-brokerDeduplicationProducerInactivityTimeoutMinutes=360
-
-# When a namespace is created without specifying the number of bundle, this
-# value will be used as the default
-defaultNumberOfNamespaceBundles=4
-
-# Max number of topics allowed to be created in the namespace. When the topics reach the max topics of the namespace,
-# the broker should reject the new topic request(include topic auto-created by the producer or consumer)
-# until the number of connected consumers decrease.
-# Using a value of 0, is disabling maxTopicsPerNamespace-limit check.
-maxTopicsPerNamespace=0
-
-# Allow schema to be auto updated at broker level. User can override this by
-# 'is_allow_auto_update_schema' of namespace policy.
-isAllowAutoUpdateSchemaEnabled=true
-
-# Enable check for minimum allowed client library version
-clientLibraryVersionCheckEnabled=false
-
-# Path for the file used to determine the rotation status for the broker when responding
-# to service discovery health checks
-statusFilePath=/usr/local/apache/htdocs
-
-# Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending
-# messages to consumer once, this limit reaches until consumer starts acknowledging messages back
-# Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction
-maxUnackedMessagesPerConsumer=50000
-
-# Max number of unacknowledged messages allowed per shared subscription. Broker will stop dispatching messages to
-# all consumers of the subscription once this limit reaches until consumer starts acknowledging messages back and
-# unack count reaches to limit/2. Using a value of 0, is disabling unackedMessage-limit
-# check and dispatcher can dispatch messages without any restriction
-maxUnackedMessagesPerSubscription=200000
-
-# Max number of unacknowledged messages allowed per broker. Once this limit reaches, broker will stop dispatching
-# messages to all shared subscription which has higher number of unack messages until subscriptions start
-# acknowledging messages back and unack count reaches to limit/2. Using a value of 0, is disabling
-# unackedMessage-limit check and broker doesn't block dispatchers
-maxUnackedMessagesPerBroker=0
-
-# Once broker reaches maxUnackedMessagesPerBroker limit, it blocks subscriptions which has higher unacked messages
-# than this percentage limit and subscription will not receive any new messages until that subscription acks back
-# limit/2 messages
-maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16
-
-# Tick time to schedule task that checks topic publish rate limiting across all topics
-# Reducing to lower value can give more accuracy while throttling publish but
-# it uses more CPU to perform frequent check. (Disable publish throttling with value 0)
-topicPublisherThrottlingTickTimeMillis=2
-
-# Enable precise rate limit for topic publish
-preciseTopicPublishRateLimiterEnable=false
-
-# Tick time to schedule task that checks broker publish rate limiting across all topics
-# Reducing to lower value can give more accuracy while throttling publish but
-# it uses more CPU to perform frequent check. (Disable publish throttling with value 0)
-brokerPublisherThrottlingTickTimeMillis=50
-
-# Max Rate(in 1 seconds) of Message allowed to publish for a broker if broker publish rate limiting enabled
-# (Disable message rate limit with value 0)
-brokerPublisherThrottlingMaxMessageRate=0
-
-# Max Rate(in 1 seconds) of Byte allowed to publish for a broker if broker publish rate limiting enabled
-# (Disable byte rate limit with value 0)
-brokerPublisherThrottlingMaxByteRate=0
-
-# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default
-# message dispatch-throttling
-dispatchThrottlingRatePerTopicInMsg=0
-
-# Default bytes per second dispatch throttling-limit for every topic. Using a value of 0, is disabling
-# default message-byte dispatch-throttling
-dispatchThrottlingRatePerTopicInByte=0
-
-# Apply dispatch rate limiting on batch message instead individual
-# messages with in batch message. (Default is disabled)
-dispatchThrottlingOnBatchMessageEnabled=false
-
-# Dispatch rate-limiting relative to publish rate.
-# (Enabling flag will make broker to dynamically update dispatch-rate relatively to publish-rate:
-# throttle-dispatch-rate = (publish-rate + configured dispatch-rate).
-dispatchThrottlingRateRelativeToPublishRate=false
-
-# By default we enable dispatch-throttling for both caught up consumers as well as consumers who have
-# backlog.
-dispatchThrottlingOnNonBacklogConsumerEnabled=true
-
-# The read failure backoff initial time in milliseconds. By default it is 15s.
-dispatcherReadFailureBackoffInitialTimeInMs=15000
-
-# The read failure backoff max time in milliseconds. By default it is 60s.
-dispatcherReadFailureBackoffMaxTimeInMs=60000
-
-# The read failure backoff mandatory stop time in milliseconds. By default it is 0s.
-dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
-
-# Precise dispathcer flow control according to history message number of each entry
-preciseDispatcherFlowControl=false
-
-# Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic
-maxConcurrentLookupRequest=50000
-
-# Max number of concurrent topic loading request broker allows to control number of zk-operations
-maxConcurrentTopicLoadRequest=5000
-
-# Max concurrent non-persistent message can be processed per connection
-maxConcurrentNonPersistentMessagePerConnection=1000
-
-# Number of worker threads to serve non-persistent topic
-numWorkerThreadsForNonPersistentTopic=8
-
-# Enable broker to load persistent topics
-enablePersistentTopics=true
-
-# Enable broker to load non-persistent topics
-enableNonPersistentTopics=true
-
-# Max number of producers allowed to connect to topic. Once this limit reaches, Broker will reject new producers
-# until the number of connected producers decrease.
-# Using a value of 0, is disabling maxProducersPerTopic-limit check.
-maxProducersPerTopic=0
-
-# Max number of producers with the same IP address allowed to connect to topic.
-# Once this limit reaches, Broker will reject new producers until the number of
-# connected producers with the same IP address decrease.
-# Using a value of 0, is disabling maxSameAddressProducersPerTopic-limit check.
-maxSameAddressProducersPerTopic=0
-
-# Enforce producer to publish encrypted messages.(default disable).
-encryptionRequireOnProducer=false
-
-# Max number of consumers allowed to connect to topic. Once this limit reaches, Broker will reject new consumers
-# until the number of connected consumers decrease.
-# Using a value of 0, is disabling maxConsumersPerTopic-limit check.
-maxConsumersPerTopic=0
-
-# Max number of consumers with the same IP address allowed to connect to topic.
-# Once this limit reaches, Broker will reject new consumers until the number of
-# connected consumers with the same IP address decrease.
-# Using a value of 0, is disabling maxSameAddressConsumersPerTopic-limit check.
-maxSameAddressConsumersPerTopic=0
-
-# Max number of subscriptions allowed to subscribe to topic. Once this limit reaches, broker will reject
-# new subscription until the number of subscribed subscriptions decrease.
-# Using a value of 0, is disabling maxSubscriptionsPerTopic limit check.
-maxSubscriptionsPerTopic=0
-
-# Max number of consumers allowed to connect to subscription. Once this limit reaches, Broker will reject new consumers
-# until the number of connected consumers decrease.
-# Using a value of 0, is disabling maxConsumersPerSubscription-limit check.
-maxConsumersPerSubscription=0
-
-# Max number of partitions per partitioned topic
-# Use 0 or negative number to disable the check
-maxNumPartitionsPerPartitionedTopic=0
-
-### --- Metadata Store --- ###
-
-# Whether we should enable metadata operations batching
-metadataStoreBatchingEnabled=true
-
-# Maximum delay to impose on batching grouping
-metadataStoreBatchingMaxDelayMillis=5
-
-# Maximum number of operations to include in a singular batch
-metadataStoreBatchingMaxOperations=1000
-
-# Maximum size of a batch
-metadataStoreBatchingMaxSizeKb=128
-
-### --- TLS --- ###
-# Deprecated - Use webServicePortTls and brokerServicePortTls instead
-tlsEnabled=false
-
-# Tls cert refresh duration in seconds (set 0 to check on every new connection)
-tlsCertRefreshCheckDurationSec=300
-
-# Path for the TLS certificate file
-tlsCertificateFilePath=
-
-# Path for the TLS private key file
-tlsKeyFilePath=
-
-# Path for the trusted TLS certificate file.
-# This cert is used to verify that any certs presented by connecting clients
-# are signed by a certificate authority. If this verification
-# fails, then the certs are untrusted and the connections are dropped.
-tlsTrustCertsFilePath=
-
-# Accept untrusted TLS certificate from client.
-# If true, a client with a cert which cannot be verified with the
-# 'tlsTrustCertsFilePath' cert will allowed to connect to the server,
-# though the cert will not be used for client authentication.
-tlsAllowInsecureConnection=false
-
-# Specify the tls protocols the broker will use to negotiate during TLS handshake
-# (a comma-separated list of protocol names).
-# Examples:- [TLSv1.3, TLSv1.2]
-tlsProtocols=
-
-# Specify the tls cipher the broker will use to negotiate during TLS Handshake
-# (a comma-separated list of ciphers).
-# Examples:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
-tlsCiphers=
-
-# Trusted client certificates are required for to connect TLS
-# Reject the Connection if the Client Certificate is not trusted.
-# In effect, this requires that all connecting clients perform TLS client
-# authentication.
-tlsRequireTrustedClientCertOnConnect=false
-
-### --- KeyStore TLS config variables --- ###
-# Enable TLS with KeyStore type configuration in broker.
-tlsEnabledWithKeyStore=false
-
-# TLS Provider for KeyStore type
-tlsProvider=
-
-# TLS KeyStore type configuration in broker: JKS, PKCS12
-tlsKeyStoreType=JKS
-
-# TLS KeyStore path in broker
-tlsKeyStore=
-
-# TLS KeyStore password for broker
-tlsKeyStorePassword=
-
-# TLS TrustStore type configuration in broker: JKS, PKCS12
-tlsTrustStoreType=JKS
-
-# TLS TrustStore path in broker
-tlsTrustStore=
-
-# TLS TrustStore password for broker
-tlsTrustStorePassword=
-
-# Whether internal client use KeyStore type to authenticate with Pulsar brokers
-brokerClientTlsEnabledWithKeyStore=false
-
-# The TLS Provider used by internal client to authenticate with other Pulsar brokers
-brokerClientSslProvider=
-
-# TLS TrustStore type configuration for internal client: JKS, PKCS12
-# used by the internal client to authenticate with Pulsar brokers
-brokerClientTlsTrustStoreType=JKS
-
-# TLS TrustStore path for internal client
-# used by the internal client to authenticate with Pulsar brokers
-brokerClientTlsTrustStore=
-
-# TLS TrustStore password for internal client,
-# used by the internal client to authenticate with Pulsar brokers
-brokerClientTlsTrustStorePassword=
-
-# Specify the tls cipher the internal client will use to negotiate during TLS Handshake
-# (a comma-separated list of ciphers)
-# e.g. [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256].
-# used by the internal client to authenticate with Pulsar brokers
-brokerClientTlsCiphers=
-
-# Specify the tls protocols the broker will use to negotiate during TLS handshake
-# (a comma-separated list of protocol names).
-# e.g. [TLSv1.3, TLSv1.2]
-# used by the internal client to authenticate with Pulsar brokers
-brokerClientTlsProtocols=
-
-# Enable or disable system topic
-systemTopicEnabled=true
-
-# Enable or disable topic level policies, topic level policies depends on the system topic
-# Please enable the system topic first.
-topicLevelPoliciesEnabled=false
-
-# If a topic remains fenced for this number of seconds, it will be closed forcefully.
-# If it is set to 0 or a negative number, the fenced topic will not be closed.
-topicFencingTimeoutSeconds=0
-
-### --- Authentication --- ###
-# Role names that are treated as "proxy roles". If the broker sees a request with
-#role as proxyRoles - it will demand to see a valid original principal.
-proxyRoles=
-
-# If this flag is set then the broker authenticates the original Auth data
-# else it just accepts the originalPrincipal and authorizes it (if required).
-authenticateOriginalAuthData=false
-
-# Enable authentication
-authenticationEnabled=false
-
-# Authentication provider name list, which is comma separated list of class names
-authenticationProviders=
-
-# Enforce authorization
-authorizationEnabled=false
-
-# Authorization provider fully qualified class-name
-authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
-
-# Allow wildcard matching in authorization
-# (wildcard matching only applicable if wildcard-char:
-# * presents at first or last position eg: *.pulsar.service, pulsar.service.*)
-authorizationAllowWildcardsMatching=false
-
-# Role names that are treated as "super-user", meaning they will be able to do all admin
-# operations and publish/consume from all topics
-superUserRoles=
-
-# Authentication settings of the broker itself. Used when the broker connects to other brokers,
-# either in same or other clusters
-brokerClientAuthenticationPlugin=
-brokerClientAuthenticationParameters=
-
-# Supported Athenz provider domain names(comma separated) for authentication
-athenzDomainNames=
-
-# When this parameter is not empty, unauthenticated users perform as anonymousUserRole
-anonymousUserRole=
-
-
-### --- Token Authentication Provider --- ###
-
-## Symmetric key
-# Configure the secret key to be used to validate auth tokens
-# The key can be specified like:
-# tokenSecretKey=data:;base64,xxxxxxxxx
-# tokenSecretKey=file:///my/secret.key ( Note: key file must be DER-encoded )
-tokenSecretKey=
-
-## Asymmetric public/private key pair
-# Configure the public key to be used to validate auth tokens
-# The key can be specified like:
-# tokenPublicKey=data:;base64,xxxxxxxxx
-# tokenPublicKey=file:///my/public.key ( Note: key file must be DER-encoded )
-tokenPublicKey=
-
-
-# The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
-tokenAuthClaim=
-
-# The token audience "claim" name, e.g. "aud", that will be used to get the audience from token.
-# If not set, audience will not be verified.
-tokenAudienceClaim=
-
-# The token audience stands for this broker. The field `tokenAudienceClaim` of a valid token, need contains this.
-tokenAudience=
-
-### --- BookKeeper Client --- ###
-
-# Authentication plugin to use when connecting to bookies
-bookkeeperClientAuthenticationPlugin=
-
-# BookKeeper auth plugin implementation specifics parameters name and values
-bookkeeperClientAuthenticationParametersName=
-bookkeeperClientAuthenticationParameters=
-
-# Timeout for BK add / read operations
-bookkeeperClientTimeoutInSeconds=30
-
-# Number of BookKeeper client worker threads
-# Default is Runtime.getRuntime().availableProcessors()
-bookkeeperClientNumWorkerThreads=
-
-# Speculative reads are initiated if a read request doesn't complete within a certain time
-# Using a value of 0, is disabling the speculative reads
-bookkeeperClientSpeculativeReadTimeoutInMillis=0
-
-# Number of channels per bookie
-bookkeeperNumberOfChannelsPerBookie=16
-
-# Enable bookies health check. Bookies that have more than the configured number of failure within
-# the interval will be quarantined for some time. During this period, new ledgers won't be created
-# on these bookies
-bookkeeperClientHealthCheckEnabled=true
-bookkeeperClientHealthCheckIntervalSeconds=60
-bookkeeperClientHealthCheckErrorThresholdPerInterval=5
-bookkeeperClientHealthCheckQuarantineTimeInSeconds=1800
-
-#bookie quarantine ratio to avoid all clients quarantine the high pressure bookie servers at the same time
-bookkeeperClientQuarantineRatio=1.0
-
-# Enable rack-aware bookie selection policy. BK will chose bookies from different racks when
-# forming a new bookie ensemble
-# This parameter related to ensemblePlacementPolicy in conf/bookkeeper.conf, if enabled, ensemblePlacementPolicy
-# should be set to org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy
-bookkeeperClientRackawarePolicyEnabled=true
-
-# Enable region-aware bookie selection policy. BK will chose bookies from
-# different regions and racks when forming a new bookie ensemble.
-# If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
-# This parameter related to ensemblePlacementPolicy in conf/bookkeeper.conf, if enabled, ensemblePlacementPolicy
-# should be set to org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy
-bookkeeperClientRegionawarePolicyEnabled=false
-
-# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
-# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
-bookkeeperClientMinNumRacksPerWriteQuorum=1
-
-# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
-# racks for a writeQuorum.
-# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
-bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
-
-# Enable/disable reordering read sequence on reading entries.
-bookkeeperClientReorderReadSequenceEnabled=false
-
-# Enable bookie isolation by specifying a list of bookie groups to choose from. Any bookie
-# outside the specified groups will not be used by the broker
-bookkeeperClientIsolationGroups=
-
-# Enable bookie secondary-isolation group if bookkeeperClientIsolationGroups doesn't
-# have enough bookie available.
-bookkeeperClientSecondaryIsolationGroups=
-
-# Minimum bookies that should be available as part of bookkeeperClientIsolationGroups
-# else broker will include bookkeeperClientSecondaryIsolationGroups bookies in isolated list.
-bookkeeperClientMinAvailableBookiesInIsolationGroups=
-
-# Set the client security provider factory class name.
-# Default: org.apache.bookkeeper.tls.TLSContextFactory
-bookkeeperTLSProviderFactoryClass=org.apache.bookkeeper.tls.TLSContextFactory
-
-# Enable tls authentication with bookie
-bookkeeperTLSClientAuthentication=false
-
-# Supported type: PEM, JKS, PKCS12. Default value: PEM
-bookkeeperTLSKeyFileType=PEM
-
-#Supported type: PEM, JKS, PKCS12. Default value: PEM
-bookkeeperTLSTrustCertTypes=PEM
-
-# Path to file containing keystore password, if the client keystore is password protected.
-bookkeeperTLSKeyStorePasswordPath=
-
-# Path to file containing truststore password, if the client truststore is password protected.
-bookkeeperTLSTrustStorePasswordPath=
-
-# Path for the TLS private key file
-bookkeeperTLSKeyFilePath=
-
-# Path for the TLS certificate file
-bookkeeperTLSCertificateFilePath=
-
-# Path for the trusted TLS certificate file
-bookkeeperTLSTrustCertsFilePath=
-
-# Enable/disable disk weight based placement. Default is false
-bookkeeperDiskWeightBasedPlacementEnabled=false
-
-# Set the interval to check the need for sending an explicit LAC
-# A value of '0' disables sending any explicit LACs. Default is 0.
-bookkeeperExplicitLacIntervalInMills=0
-
-# Use older Bookkeeper wire protocol with bookie
-bookkeeperUseV2WireProtocol=true
-
-# Expose bookkeeper client managed ledger stats to prometheus. default is false
-# bookkeeperClientExposeStatsToPrometheus=false
-
-### --- Managed Ledger --- ###
-
-# Number of bookies to use when creating a ledger
-managedLedgerDefaultEnsembleSize=1
-
-# Number of copies to store for each message
-managedLedgerDefaultWriteQuorum=1
-
-# Number of guaranteed copies (acks to wait before write is complete)
-managedLedgerDefaultAckQuorum=1
-
-# How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds).
-# Default is 60 seconds
-managedLedgerCursorPositionFlushSeconds=60
-
-# Default type of checksum to use when writing to BookKeeper. Default is "CRC32C"
-# Other possible options are "CRC32", "MAC" or "DUMMY" (no checksum).
-managedLedgerDigestType=CRC32C
-
-# Number of threads to be used for managed ledger tasks dispatching
-managedLedgerNumWorkerThreads=4
-
-# Number of threads to be used for managed ledger scheduled tasks
-managedLedgerNumSchedulerThreads=4
-
-# Amount of memory to use for caching data payload in managed ledger. This memory
-# is allocated from JVM direct memory and it's shared across all the topics
-# running in the same broker. By default, uses 1/5th of available direct memory
-managedLedgerCacheSizeMB=
-
-# Whether we should make a copy of the entry payloads when inserting in cache
-managedLedgerCacheCopyEntries=false
-
-# Threshold to which bring down the cache level when eviction is triggered
-managedLedgerCacheEvictionWatermark=0.9
-
-# Configure the cache eviction frequency for the managed ledger cache (evictions/sec)
-managedLedgerCacheEvictionFrequency=100.0
-
-# All entries that have stayed in cache for more than the configured time, will be evicted
-managedLedgerCacheEvictionTimeThresholdMillis=1000
-
-# Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged'
-# and thus should be set as inactive.
-managedLedgerCursorBackloggedThreshold=1000
-
-# Rate limit the amount of writes generated by consumer acking the messages
-managedLedgerDefaultMarkDeleteRateLimit=0.1
-
-# Max number of entries to append to a ledger before triggering a rollover
-# A ledger rollover is triggered after the min rollover time has passed
-# and one of the following conditions is true:
-# * The max rollover time has been reached
-# * The max entries have been written to the ledger
-# * The max ledger size has been written to the ledger
-managedLedgerMaxEntriesPerLedger=50000
-
-# Minimum time between ledger rollover for a topic
-managedLedgerMinLedgerRolloverTimeMinutes=10
-
-# Maximum time before forcing a ledger rollover for a topic
-managedLedgerMaxLedgerRolloverTimeMinutes=240
-
-# Max number of entries to append to a cursor ledger
-managedLedgerCursorMaxEntriesPerLedger=50000
-
-# Max time before triggering a rollover on a cursor ledger
-managedLedgerCursorRolloverTimeInSeconds=14400
-
-# Maximum ledger size before triggering a rollover for a topic (MB)
-managedLedgerMaxSizePerLedgerMbytes=2048
-
-# Max number of "acknowledgment holes" that are going to be persistently stored.
-# When acknowledging out of order, a consumer will leave holes that are supposed
-# to be quickly filled by acking all the messages. The information of which
-# messages are acknowledged is persisted by compressing in "ranges" of messages
-# that were acknowledged. After the max number of ranges is reached, the information
-# will only be tracked in memory and messages will be redelivered in case of
-# crashes.
-managedLedgerMaxUnackedRangesToPersist=10000
-
-# Max number of "acknowledgment holes" that can be stored in Zookeeper. If number of unack message range is higher
-# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
-# zookeeper.
-managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000
-
-# Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
-# corrupted at bookkeeper and managed-cursor is stuck at that ledger.
-autoSkipNonRecoverableData=false
-
-# operation timeout while updating managed-ledger metadata.
-managedLedgerMetadataOperationsTimeoutSeconds=60
-
-# Read entries timeout when broker tries to read messages from bookkeeper.
-managedLedgerReadEntryTimeoutSeconds=0
-
-# Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
-managedLedgerAddEntryTimeoutSeconds=0
-
-# New entries check delay for the cursor under the managed ledger.
-# If no new messages in the topic, the cursor will try to check again after the delay time.
-# For consumption latency sensitive scenario, can set to a smaller value or set to 0.
-# Of course, use a smaller value may degrade consumption throughput. Default is 10ms.
-managedLedgerNewEntriesCheckDelayInMillis=10
-
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
-# Managed ledger prometheus stats latency rollover seconds (default: 60s)
-managedLedgerPrometheusStatsLatencyRolloverSeconds=60
-
-# Whether trace managed ledger task execution time
-managedLedgerTraceTaskExecution=true
-
-# If you want to custom bookie ID or use a dynamic network address for the bookie,
-# you can set this option.
-# Bookie advertises itself using bookieId rather than
-# BookieSocketAddress (hostname:port or IP:port).
-# bookieId is a non empty string that can contain ASCII digits and letters ([a-zA-Z9-0]),
-# colons, dashes, and dots.
-# For more information about bookieId, see http://bookkeeper.apache.org/bps/BP-41-bookieid/.
-# bookieId=
-
-### --- Load balancer --- ###
-
-loadManagerClassName=org.apache.pulsar.broker.loadbalance.NoopLoadManager
-
-# Enable load balancer
-loadBalancerEnabled=false
-
-# Percentage of change to trigger load report update
-loadBalancerReportUpdateThresholdPercentage=10
-
-# maximum interval to update load report
-loadBalancerReportUpdateMaxIntervalMinutes=15
-
-# Frequency of report to collect
-loadBalancerHostUsageCheckIntervalMinutes=1
-
-# Load shedding interval. Broker periodically checks whether some traffic should be offload from
-# some over-loaded broker to other under-loaded brokers
-loadBalancerSheddingIntervalMinutes=1
-
-# Prevent the same topics to be shed and moved to other broker more than once within this timeframe
-loadBalancerSheddingGracePeriodMinutes=30
-
-# Usage threshold to allocate max number of topics to broker
-loadBalancerBrokerMaxTopics=50000
-
-# Interval to flush dynamic resource quota to ZooKeeper
-loadBalancerResourceQuotaUpdateIntervalMinutes=15
-
-# enable/disable namespace bundle auto split
-loadBalancerAutoBundleSplitEnabled=true
-
-# enable/disable automatic unloading of split bundles
-loadBalancerAutoUnloadSplitBundlesEnabled=true
-
-# maximum topics in a bundle, otherwise bundle split will be triggered
-loadBalancerNamespaceBundleMaxTopics=1000
-
-# maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
-loadBalancerNamespaceBundleMaxSessions=1000
-
-# maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered
-loadBalancerNamespaceBundleMaxMsgRate=30000
-
-# maximum bandwidth (in + out) in a bundle, otherwise bundle split will be triggered
-loadBalancerNamespaceBundleMaxBandwidthMbytes=100
-
-# maximum number of bundles in a namespace
-loadBalancerNamespaceMaximumBundles=128
-
-# The broker resource usage threshold.
-# When the broker resource usage is greater than the pulsar cluster average resource usage,
-# the threshold shedder will be triggered to offload bundles from the broker.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerBrokerThresholdShedderPercentage=10
-
-# When calculating new resource usage, the history usage accounts for.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerHistoryResourcePercentage=0.9
-
-# The BandWithIn usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerBandwithInResourceWeight=1.0
-
-# The BandWithOut usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerBandwithOutResourceWeight=1.0
-
-# The CPU usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerCPUResourceWeight=1.0
-
-# The heap memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerMemoryResourceWeight=1.0
-
-# The direct memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerDirectMemoryResourceWeight=1.0
-
-# Bundle unload minimum throughput threshold (MB), avoiding bundle unload frequently.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerBundleUnloadMinThroughputThreshold=10
-
-# Time to wait for the unloading of a namespace bundle
-namespaceBundleUnloadingTimeoutMs=60000
-
-### --- Replication --- ###
-
-# Enable replication metrics
-replicationMetricsEnabled=true
-
-# Max number of connections to open for each broker in a remote cluster
-# More connections host-to-host lead to better throughput over high-latency
-# links.
-replicationConnectionsPerBroker=16
-
-# Replicator producer queue size
-replicationProducerQueueSize=1000
-
-# Duration to check replication policy to avoid replicator inconsistency
-# due to missing ZooKeeper watch (disable with value 0)
-replicationPolicyCheckDurationSeconds=600
-
-# Default message retention time
-defaultRetentionTimeInMinutes=0
-
-# Default retention size
-defaultRetentionSizeInMB=0
-
-# How often to check whether the connections are still alive
-keepAliveIntervalSeconds=30
-
-### --- WebSocket --- ###
-
-# Enable the WebSocket API service in broker
-webSocketServiceEnabled=true
-
-# Number of IO threads in Pulsar Client used in WebSocket proxy
-webSocketNumIoThreads=8
-
-# Number of connections per Broker in Pulsar Client used in WebSocket proxy
-webSocketConnectionsPerBroker=8
-
-# Time in milliseconds that idle WebSocket session times out
-webSocketSessionIdleTimeoutMillis=300000
-
-# The maximum size of a text message during parsing in WebSocket proxy
-webSocketMaxTextFrameSize=1048576
-
-### --- Metrics --- ###
-
-# Enable topic level metrics
-exposeTopicLevelMetricsInPrometheus=true
-
-# Time in milliseconds that metrics endpoint would time out. Default is 30s.
-# Increase it if there are a lot of topics to expose topic-level metrics.
-# Set it to 0 to disable timeout.
-metricsServletTimeoutMs=30000
-
-# Classname of Pluggable JVM GC metrics logger that can log GC specific metrics
-# jvmGCMetricsLoggerClassName=
-
-### --- Broker Web Stats --- ###
-
-# Enable topic level metrics
-exposePublisherStats=true
-
-# Enable expose the precise backlog stats.
-# Set false to use published counter and consumed counter to calculate, this would be more efficient but may be inaccurate.
-# Default is false.
-exposePreciseBacklogInPrometheus=false
-
-# Enable splitting topic and partition label in Prometheus.
-# If enabled, a topic name will split into 2 parts, one is topic name without partition index,
-# another one is partition index, e.g. (topic=xxx, partition=0).
-# If the topic is a non-partitioned topic, -1 will be used for the partition index.
-# If disabled, one label to represent the topic and partition, e.g. (topic=xxx-partition-0)
-# Default is false.
-
-splitTopicAndPartitionLabelInPrometheus=false
-
-### --- Deprecated config variables --- ###
-
-# Deprecated. Use configurationStoreServers
-globalZookeeperServers=
-
-# Deprecated. Use brokerDeleteInactiveTopicsFrequencySeconds
-brokerServicePurgeInactiveFrequencyInSeconds=60
-
-### --- BookKeeper Configuration --- #####
-
-ledgerStorageClass=org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage
-
-# The maximum netty frame size in bytes. Any message received larger than this will be rejected. The default value is 5MB.
-nettyMaxFrameSizeBytes=5253120
-
-# Size of Write Cache. Memory is allocated from JVM direct memory.
-# Write cache is used to buffer entries before flushing into the entry log
-# For good performance, it should be big enough to hold a substantial amount
-# of entries in the flush interval
-# By default it will be allocated to 1/4th of the available direct memory
-dbStorage_writeCacheMaxSizeMb=
-
-# Size of Read cache. Memory is allocated from JVM direct memory.
-# This read cache is pre-filled doing read-ahead whenever a cache miss happens
-# By default it will be allocated to 1/4th of the available direct memory
-dbStorage_readAheadCacheMaxSizeMb=
-
-# How many entries to pre-fill in cache after a read cache miss
-dbStorage_readAheadCacheBatchSize=1000
-
-flushInterval=60000
-
-## RocksDB specific configurations
-## DbLedgerStorage uses RocksDB to store the indexes from
-## (ledgerId, entryId) -> (entryLog, offset)
-
-# Size of RocksDB block-cache. For best performance, this cache
-# should be big enough to hold a significant portion of the index
-# database which can reach ~2GB in some cases
-# Default is to use 10% of the direct memory size
-dbStorage_rocksDB_blockCacheSize=
-
-# Other RocksDB specific tunables
-dbStorage_rocksDB_writeBufferSizeMB=4
-dbStorage_rocksDB_sstSizeInMB=4
-dbStorage_rocksDB_blockSize=4096
-dbStorage_rocksDB_bloomFilterBitsPerKey=10
-dbStorage_rocksDB_numLevels=-1
-dbStorage_rocksDB_numFilesInLevel0=4
-dbStorage_rocksDB_maxSizeInLevel1MB=256
-
-# Maximum latency to impose on a journal write to achieve grouping
-journalMaxGroupWaitMSec=1
-
-# Should the data be fsynced on journal before acknowledgment.
-journalSyncData=false
-
-
-# For each ledger dir, maximum disk space which can be used.
-# Default is 0.95f. i.e. 95% of disk can be used at most after which nothing will
-# be written to that partition. If all ledger dir partions are full, then bookie
-# will turn to readonly mode if 'readOnlyModeEnabled=true' is set, else it will
-# shutdown.
-# Valid values should be in between 0 and 1 (exclusive).
-diskUsageThreshold=0.99
-
-# The disk free space low water mark threshold.
-# Disk is considered full when usage threshold is exceeded.
-# Disk returns back to non-full state when usage is below low water mark threshold.
-# This prevents it from going back and forth between these states frequently
-# when concurrent writes and compaction are happening. This also prevent bookie from
-# switching frequently between read-only and read-writes states in the same cases.
-diskUsageWarnThreshold=0.99
-
-# Whether the bookie allowed to use a loopback interface as its primary
-# interface(i.e. the interface it uses to establish its identity)?
-# By default, loopback interfaces are not allowed as the primary
-# interface.
-# Using a loopback interface as the primary interface usually indicates
-# a configuration error. For example, its fairly common in some VPS setups
-# to not configure a hostname, or to have the hostname resolve to
-# 127.0.0.1. If this is the case, then all bookies in the cluster will
-# establish their identities as 127.0.0.1:3181, and only one will be able
-# to join the cluster. For VPSs configured like this, you should explicitly
-# set the listening interface.
-allowLoopback=true
-
-# How long the interval to trigger next garbage collection, in milliseconds
-# Since garbage collection is running in background, too frequent gc
-# will heart performance. It is better to give a higher number of gc
-# interval if there is enough disk capacity.
-gcWaitTime=300000
-
-# Enable topic auto creation if new producer or consumer connected (disable auto creation with value false)
-allowAutoTopicCreation=true
-
-# The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
-allowAutoTopicCreationType=non-partitioned
-
-# Enable subscription auto creation if new consumer connected (disable auto creation with value false)
-allowAutoSubscriptionCreation=true
-
-# The number of partitioned topics that is allowed to be automatically created if allowAutoTopicCreationType is partitioned.
-defaultNumPartitions=1
-
-### --- Transaction config variables --- ###
-# Enable transaction coordinator in broker
-transactionCoordinatorEnabled=true
-transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.InMemTransactionMetadataStoreProvider
-
-# Transaction buffer take snapshot transaction count
-transactionBufferSnapshotMaxTransactionCount=1000
-
-# Transaction buffer take snapshot interval time
-# Unit : millisecond
-transactionBufferSnapshotMinTimeInMillis=5000
-
-### --- Packages management service configuration variables (begin) --- ###
-
-# Enable the packages management service or not
-enablePackagesManagement=false
-
-# The packages management service storage service provide
-packagesManagementStorageProvider=org.apache.pulsar.packages.management.storage.bookkeeper.BookKeeperPackagesStorageProvider
-
-# When the packages storage provider is bookkeeper, you can use this configuration to
-# control the number of replicas for storing the package
-packagesReplicas=1
-
-# The bookkeeper ledger root path
-packagesManagementLedgerRootPath=/ledgers
-
-### --- Packages management service configuration variables (end) --- ###
[flink] 04/05: [FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6075746cb3bf4462cdd7d474c96b65e0bdc57d99
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Sep 28 15:28:14 2022 +0800
[FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
---
.../f4d91193-72ba-4ce4-ad83-98f780dce581 | 6 +
flink-connectors/flink-connector-pulsar/pom.xml | 2 -
.../source/enumerator/topic/TopicPartition.java | 5 +
.../connector/pulsar/sink/PulsarSinkITCase.java | 155 +++++++++++++--------
.../pulsar/source/PulsarUnorderedSourceITCase.java | 14 +-
.../pulsar/testutils/PulsarTestCommonUtils.java | 5 +-
.../pulsar/testutils/PulsarTestContext.java | 6 +-
.../testutils/runtime/PulsarRuntimeOperator.java | 15 +-
.../runtime/container/PulsarContainerRuntime.java | 25 ++--
.../testutils/runtime/mock/PulsarMockRuntime.java | 5 +-
.../testutils/sink/PulsarPartitionDataReader.java | 93 +++++++++++++
.../testutils/sink/PulsarSinkTestContext.java | 124 +++++++++++++++++
.../testutils/sink/PulsarSinkTestSuiteBase.java | 37 +++++
.../source}/KeyedPulsarPartitionDataWriter.java | 15 +-
.../source/UnorderedSourceTestSuiteBase.java | 5 +-
.../cases}/KeySharedSubscriptionContext.java | 24 ++--
.../cases/MultipleTopicConsumingContext.java | 9 +-
.../source/cases}/SharedSubscriptionContext.java | 12 +-
.../src/test/resources/docker/bootstrap.sh | 28 ----
...norderedE2ECase.java => PulsarSinkE2ECase.java} | 31 ++---
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 4 +-
.../FlinkContainerWithPulsarEnvironment.java | 5 +-
22 files changed, 435 insertions(+), 190 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/archunit-violations/f4d91193-72ba-4ce4-ad83-98f780dce581 b/flink-connectors/flink-connector-pulsar/archunit-violations/f4d91193-72ba-4ce4-ad83-98f780dce581
index 03d0ae583af..40e7dc9686c 100644
--- a/flink-connectors/flink-connector-pulsar/archunit-violations/f4d91193-72ba-4ce4-ad83-98f780dce581
+++ b/flink-connectors/flink-connector-pulsar/archunit-violations/f4d91193-72ba-4ce4-ad83-98f780dce581
@@ -1,3 +1,9 @@
+org.apache.flink.connector.pulsar.sink.PulsarSinkITCase does not satisfy: only one of the following predicates match:\
+* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
+* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
+* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
+* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
+ or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
org.apache.flink.connector.pulsar.source.PulsarSourceITCase does not satisfy: only one of the following predicates match:\
* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
diff --git a/flink-connectors/flink-connector-pulsar/pom.xml b/flink-connectors/flink-connector-pulsar/pom.xml
index 49b06b26806..ef0a2ffc12a 100644
--- a/flink-connectors/flink-connector-pulsar/pom.xml
+++ b/flink-connectors/flink-connector-pulsar/pom.xml
@@ -320,7 +320,6 @@ under the License.
<include>**/testutils/**</include>
<include>META-INF/LICENSE</include>
<include>META-INF/NOTICE</include>
- <include>containers/txnStandalone.conf</include>
</includes>
</configuration>
</execution>
@@ -344,7 +343,6 @@ under the License.
<include>**/testutils/**</include>
<include>META-INF/LICENSE</include>
<include>META-INF/NOTICE</include>
- <include>containers/txnStandalone.conf</include>
</includes>
</configuration>
</execution>
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
index b3035cde848..1113771b76d 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
@@ -29,6 +29,7 @@ import java.util.Objects;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.createFullRange;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -58,6 +59,10 @@ public class TopicPartition implements Serializable {
*/
private final TopicRange range;
+ public TopicPartition(String topic, int partitionId) {
+ this(topic, partitionId, createFullRange());
+ }
+
public TopicPartition(String topic, int partitionId, TopicRange range) {
this.topic = topicName(checkNotNull(topic));
this.partitionId = partitionId;
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
index 4679a994f1c..b0ef390b728 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
@@ -20,14 +20,28 @@ package org.apache.flink.connector.pulsar.sink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import org.apache.flink.connector.pulsar.testutils.function.ControlSource;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestSuiteBase;
+import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
+import org.apache.flink.connector.testframe.junit.annotations.TestContext;
+import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
+import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.runtime.minicluster.RpcServiceSharing;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.junit.FailsOnJava11;
import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
@@ -41,64 +55,91 @@ import static org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSer
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for using PulsarSink writing to a Pulsar cluster. */
-class PulsarSinkITCase extends PulsarTestSuiteBase {
-
- private static final int PARALLELISM = 1;
-
- @RegisterExtension
- private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
- new MiniClusterExtension(
- new MiniClusterResourceConfiguration.Builder()
- .setNumberTaskManagers(1)
- .setNumberSlotsPerTaskManager(PARALLELISM)
- .setRpcServiceSharing(RpcServiceSharing.DEDICATED)
- .withHaLeadershipControl()
- .build());
-
- // Using this extension for creating shared reference which would be used in source function.
- @RegisterExtension final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
-
- @ParameterizedTest
- @EnumSource(DeliveryGuarantee.class)
- void writeRecordsToPulsar(DeliveryGuarantee guarantee) throws Exception {
- // A random topic with partition 4.
- String topic = randomAlphabetic(8);
- operator().createTopic(topic, 4);
- int counts = ThreadLocalRandom.current().nextInt(100, 200);
-
- ControlSource source =
- new ControlSource(
- sharedObjects,
- operator(),
- topic,
- guarantee,
- counts,
- Duration.ofMillis(50),
- Duration.ofMinutes(5));
- PulsarSink<String> sink =
- PulsarSink.builder()
- .setServiceUrl(operator().serviceUrl())
- .setAdminUrl(operator().adminUrl())
- .setDeliveryGuarantee(guarantee)
- .setTopics(topic)
- .setProducerName("pulsar-sink-it-case")
- .setSerializationSchema(flinkSchema(new SimpleStringSchema()))
- .build();
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- env.setParallelism(PARALLELISM);
- if (guarantee != DeliveryGuarantee.NONE) {
- env.enableCheckpointing(500L);
- }
- env.addSource(source).sinkTo(sink);
- env.execute();
+@Category(value = {FailsOnJava11.class})
+class PulsarSinkITCase {
+
+ /** Integration test based on connector testing framework. */
+ @Nested
+ class IntegrationTest extends PulsarSinkTestSuiteBase {
+
+ @TestEnv MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
+
+ @TestExternalSystem
+ PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.mock());
+
+ @TestSemantics
+ CheckpointingMode[] semantics =
+ new CheckpointingMode[] {
+ CheckpointingMode.EXACTLY_ONCE, CheckpointingMode.AT_LEAST_ONCE
+ };
+
+ @TestContext
+ PulsarTestContextFactory<String, PulsarSinkTestContext> sinkContext =
+ new PulsarTestContextFactory<>(pulsar, PulsarSinkTestContext::new);
+ }
+
+ /** Tests for using PulsarSink writing to a Pulsar cluster. */
+ @Nested
+ class DeliveryGuaranteeTest extends PulsarTestSuiteBase {
- List<String> expectedRecords = source.getExpectedRecords();
- List<String> consumedRecords = source.getConsumedRecords();
+ private static final int PARALLELISM = 1;
- assertThat(consumedRecords)
- .hasSameSizeAs(expectedRecords)
- .containsExactlyInAnyOrderElementsOf(expectedRecords);
+ @RegisterExtension
+ private final MiniClusterExtension clusterExtension =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(PARALLELISM)
+ .setRpcServiceSharing(RpcServiceSharing.DEDICATED)
+ .withHaLeadershipControl()
+ .build());
+
+ // Using this extension for creating shared reference which would be used in source
+ // function.
+ @RegisterExtension
+ final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
+
+ @ParameterizedTest
+ @EnumSource(DeliveryGuarantee.class)
+ void writeRecordsToPulsar(DeliveryGuarantee guarantee) throws Exception {
+ // A random topic with partition 4.
+ String topic = randomAlphabetic(8);
+ operator().createTopic(topic, 4);
+ int counts = ThreadLocalRandom.current().nextInt(100, 200);
+
+ ControlSource source =
+ new ControlSource(
+ sharedObjects,
+ operator(),
+ topic,
+ guarantee,
+ counts,
+ Duration.ofMillis(50),
+ Duration.ofMinutes(5));
+ PulsarSink<String> sink =
+ PulsarSink.builder()
+ .setServiceUrl(operator().serviceUrl())
+ .setAdminUrl(operator().adminUrl())
+ .setDeliveryGuarantee(guarantee)
+ .setTopics(topic)
+ .setSerializationSchema(flinkSchema(new SimpleStringSchema()))
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ env.setParallelism(PARALLELISM);
+ if (guarantee != DeliveryGuarantee.NONE) {
+ env.enableCheckpointing(500L);
+ }
+ env.addSource(source).sinkTo(sink);
+ env.execute();
+
+ List<String> expectedRecords = source.getExpectedRecords();
+ List<String> consumedRecords = source.getConsumedRecords();
+
+ assertThat(consumedRecords)
+ .hasSameSizeAs(expectedRecords)
+ .containsExactlyInAnyOrderElementsOf(expectedRecords);
+ }
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
index d18c6c2633b..070a16126a7 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
@@ -22,7 +22,8 @@ import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.KeySharedSubscriptionContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.SharedSubscriptionContext;
import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
@@ -34,8 +35,6 @@ import org.apache.flink.testutils.junit.FailsOnJava11;
import org.apache.pulsar.client.api.SubscriptionType;
import org.junit.experimental.categories.Category;
-import static org.apache.pulsar.client.api.SubscriptionType.Shared;
-
/**
* Unit test class for {@link PulsarSource}. Used for {@link SubscriptionType#Shared} subscription.
*/
@@ -53,7 +52,10 @@ public class PulsarUnorderedSourceITCase extends UnorderedSourceTestSuiteBase<St
CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
@TestContext
- PulsarTestContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
- new PulsarTestContextFactory<>(
- pulsar, env -> new MultipleTopicConsumingContext(env, Shared));
+ PulsarTestContextFactory<String, SharedSubscriptionContext> sharedSubscription =
+ new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
+
+ @TestContext
+ PulsarTestContextFactory<String, KeySharedSubscriptionContext> keySharedSubscription =
+ new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
index 96b1ca62f2a..994601329d6 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.streaming.api.CheckpointingMode;
@@ -59,9 +58,7 @@ public class PulsarTestCommonUtils {
public static PulsarPartitionSplit createPartitionSplit(
String topic, int partitionId, Boundedness boundedness, MessageId latestConsumedId) {
- TopicPartition topicPartition =
- new TopicPartition(topic, partitionId, TopicRange.createFullRange());
-
+ TopicPartition topicPartition = new TopicPartition(topic, partitionId);
StopCursor stopCursor =
boundedness == Boundedness.BOUNDED ? StopCursor.latest() : StopCursor.never();
return new PulsarPartitionSplit(topicPartition, stopCursor, latestConsumedId, null);
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
index a4e44441bfd..1408656f986 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
@@ -52,10 +52,12 @@ public abstract class PulsarTestContext<T> implements ExternalContext {
@Override
public List<URL> getConnectorJarPaths() {
- // We don't need any tests jar definition. They are provided in docker-related environments.
+ // We don't need any test jars definition. They are provided in docker-related environments.
return Collections.emptyList();
}
@Override
- public void close() throws Exception {}
+ public void close() throws Exception {
+ // All the topics would be deleted in the PulsarRuntime. No need to manually close them.
+ }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
index 9787a8ceeb9..5d14d7ee081 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
@@ -22,7 +22,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.config.PulsarConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import org.apache.flink.connector.testframe.external.ExternalContext;
import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
@@ -48,7 +47,6 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import java.io.Closeable;
-import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
@@ -228,12 +226,7 @@ public class PulsarRuntimeOperator implements Closeable {
public List<TopicPartition> topicInfo(String topic) {
try {
return client().getPartitionsForTopic(topic).get().stream()
- .map(
- p ->
- new TopicPartition(
- topic,
- TopicName.getPartitionIndex(p),
- TopicRange.createFullRange()))
+ .map(p -> new TopicPartition(topic, TopicName.getPartitionIndex(p)))
.collect(toList());
} catch (InterruptedException | ExecutionException e) {
throw new IllegalStateException(e);
@@ -452,7 +445,7 @@ public class PulsarRuntimeOperator implements Closeable {
/** This method is used for test framework. You can't close this operator manually. */
@Override
- public void close() throws IOException {
+ public void close() throws PulsarClientException {
if (admin != null) {
admin.close();
}
@@ -485,7 +478,7 @@ public class PulsarRuntimeOperator implements Closeable {
}
}
- private synchronized <T> Producer<T> createProducer(String topic, Schema<T> schema) {
+ private <T> Producer<T> createProducer(String topic, Schema<T> schema) {
ProducerBuilder<T> builder =
client().newProducer(schema)
.topic(topic)
@@ -496,7 +489,7 @@ public class PulsarRuntimeOperator implements Closeable {
return sneakyClient(builder::create);
}
- private synchronized <T> Consumer<T> createConsumer(String topic, Schema<T> schema) {
+ private <T> Consumer<T> createConsumer(String topic, Schema<T> schema) {
// Create the earliest subscription if it's not existed.
List<String> subscriptions = sneakyAdmin(() -> admin().topics().getSubscriptions(topic));
if (!subscriptions.contains(SUBSCRIPTION_NAME)) {
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
index 910985338b9..eac42f063bc 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
@@ -24,7 +24,6 @@ import org.apache.flink.util.DockerImageVersions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.BindMode;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.PulsarContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -88,22 +87,30 @@ public class PulsarContainerRuntime implements PulsarRuntime {
return;
}
- // Override the default configuration in container for enabling the Pulsar transaction.
- container.withClasspathResourceMapping(
- "docker/bootstrap.sh", "/pulsar/bin/bootstrap.sh", BindMode.READ_ONLY);
- // Waiting for the Pulsar border is ready.
+ // Override the default standalone configuration by system environments.
+ container.withEnv("PULSAR_PREFIX_transactionCoordinatorEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_acknowledgmentAtBatchIndexLevelEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_systemTopicEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_brokerDeduplicationEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_defaultNumberOfNamespaceBundles", "1");
+ // Change the default bootstrap script, it will override the default configuration
+ // and start a standalone Pulsar without streaming storage and function worker.
+ container.withCommand(
+ "sh",
+ "-c",
+ "/pulsar/bin/apply-config-from-env.py /pulsar/conf/standalone.conf && /pulsar/bin/pulsar standalone --no-functions-worker -nss");
+ // Waiting for the Pulsar broker and the transaction is ready after the container started.
container.waitingFor(
- forHttp("/admin/v2/namespaces/public/default")
+ forHttp(
+ "/admin/v2/persistent/pulsar/system/transaction_coordinator_assign/partitions")
.forPort(BROKER_HTTP_PORT)
.forStatusCode(200)
.withStartupTimeout(Duration.ofMinutes(5)));
- // Set custom startup script.
- container.withCommand("sh /pulsar/bin/bootstrap.sh");
// Start the Pulsar Container.
container.start();
// Append the output to this runtime logger. Used for local debug purpose.
- container.followOutput(new Slf4jLogConsumer(LOG).withSeparateOutputStreams());
+ container.followOutput(new Slf4jLogConsumer(LOG, true));
// Create the operator.
if (boundFlink) {
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
index 326c3ff6a89..b0df4c5d1ff 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
@@ -64,10 +64,10 @@ public class PulsarMockRuntime implements PulsarRuntime {
@Override
public void tearDown() {
try {
- pulsarService.close();
if (operator != null) {
operator.close();
}
+ pulsarService.close();
} catch (Exception e) {
throw new IllegalStateException(e);
}
@@ -95,7 +95,7 @@ public class PulsarMockRuntime implements PulsarRuntime {
configuration.setAuthorizationEnabled(false);
configuration.setAllowAutoTopicCreation(true);
configuration.setBrokerDeleteInactiveTopicsEnabled(false);
-
+ configuration.setTopicLevelPoliciesEnabled(true);
configuration.setWebSocketServiceEnabled(false);
// Use runtime dynamic ports
configuration.setBrokerServicePort(Optional.of(0));
@@ -107,6 +107,7 @@ public class PulsarMockRuntime implements PulsarRuntime {
configuration.setTransactionCoordinatorEnabled(true);
configuration.setTransactionMetadataStoreProviderClassName(
"org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider");
+ configuration.setDefaultNumberOfNamespaceBundles(1);
return configuration;
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarPartitionDataReader.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarPartitionDataReader.java
new file mode 100644
index 00000000000..6ef9510d116
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarPartitionDataReader.java
@@ -0,0 +1,93 @@
+/*
+ * 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.flink.connector.pulsar.testutils.sink;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+
+/** The data reader for a specified topic partition from Pulsar. */
+public class PulsarPartitionDataReader<T> implements ExternalSystemDataReader<T>, Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PulsarPartitionDataReader.class);
+
+ private final Consumer<T> consumer;
+
+ public PulsarPartitionDataReader(
+ PulsarRuntimeOperator operator, String fullTopicName, Schema<T> schema)
+ throws PulsarClientException {
+ // Create client for supporting the use in E2E test.
+ String subscriptionName = randomAlphanumeric(12);
+ this.consumer =
+ operator.client()
+ .newConsumer(schema)
+ .topic(fullTopicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ }
+
+ @Override
+ public List<T> poll(Duration timeout) {
+ List<T> results = new ArrayList<>();
+
+ while (true) {
+ try {
+ int millis = Math.toIntExact(timeout.toMillis());
+ Message<T> message = consumer.receive(millis, MILLISECONDS);
+
+ if (message != null) {
+ consumer.acknowledgeCumulative(message);
+ results.add(message.getValue());
+ } else {
+ break;
+ }
+ } catch (Exception e) {
+ LOG.error("", e);
+ break;
+ }
+ }
+
+ return results;
+ }
+
+ @Override
+ public void close() throws PulsarClientException {
+ consumer.close();
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
new file mode 100644
index 00000000000..6b9cf2d1041
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
@@ -0,0 +1,124 @@
+/*
+ * 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.flink.connector.pulsar.testutils.sink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.sink.PulsarSink;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
+import org.apache.flink.connector.testframe.external.sink.DataStreamSinkV2ExternalContext;
+import org.apache.flink.connector.testframe.external.sink.TestingSinkSettings;
+
+import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
+
+import org.apache.pulsar.client.api.Schema;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient;
+import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_BATCHING_MAX_MESSAGES;
+import static org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema.pulsarSchema;
+import static org.apache.flink.connector.pulsar.testutils.PulsarTestCommonUtils.toDeliveryGuarantee;
+
+/** Common sink test context for pulsar based test. */
+public class PulsarSinkTestContext extends PulsarTestContext<String>
+ implements DataStreamSinkV2ExternalContext<String> {
+
+ private static final String TOPIC_NAME_PREFIX = "flink-sink-topic-";
+ private static final int RECORD_SIZE_UPPER_BOUND = 300;
+ private static final int RECORD_SIZE_LOWER_BOUND = 100;
+ private static final int RECORD_STRING_SIZE = 20;
+
+ private String topicName = topicName();
+ private final Closer closer = Closer.create();
+
+ public PulsarSinkTestContext(PulsarTestEnvironment environment) {
+ super(environment, Schema.STRING);
+ }
+
+ @Override
+ protected String displayName() {
+ return "write messages into one topic in Pulsar";
+ }
+
+ @Override
+ public Sink<String> createSink(TestingSinkSettings sinkSettings) {
+ operator.createTopic(topicName, 4);
+ DeliveryGuarantee guarantee = toDeliveryGuarantee(sinkSettings.getCheckpointingMode());
+
+ return PulsarSink.builder()
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
+ .setTopics(topicName)
+ .setDeliveryGuarantee(guarantee)
+ .setSerializationSchema(pulsarSchema(schema))
+ .enableSchemaEvolution()
+ .setConfig(PULSAR_BATCHING_MAX_MESSAGES, 4)
+ .build();
+ }
+
+ @Override
+ public ExternalSystemDataReader<String> createSinkDataReader(TestingSinkSettings sinkSettings) {
+ PulsarPartitionDataReader<String> reader =
+ sneakyClient(
+ () -> new PulsarPartitionDataReader<>(operator, topicName, Schema.STRING));
+ closer.register(reader);
+
+ return reader;
+ }
+
+ @Override
+ public List<String> generateTestData(TestingSinkSettings sinkSettings, long seed) {
+ Random random = new Random(seed);
+ int recordSize =
+ random.nextInt(RECORD_SIZE_UPPER_BOUND - RECORD_SIZE_LOWER_BOUND)
+ + RECORD_SIZE_LOWER_BOUND;
+ List<String> records = new ArrayList<>(recordSize);
+ for (int i = 0; i < recordSize; i++) {
+ int size = random.nextInt(RECORD_STRING_SIZE) + RECORD_STRING_SIZE;
+ String record = "index:" + i + "-data:" + randomAlphanumeric(size);
+ records.add(record);
+ }
+
+ return records;
+ }
+
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return Types.STRING;
+ }
+
+ @Override
+ public void close() throws Exception {
+ // Change the topic name after finishing a test case.
+ closer.register(() -> topicName = topicName());
+ closer.close();
+ }
+
+ private String topicName() {
+ return TOPIC_NAME_PREFIX + randomAlphanumeric(8);
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestSuiteBase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestSuiteBase.java
new file mode 100644
index 00000000000..0695a43569b
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestSuiteBase.java
@@ -0,0 +1,37 @@
+/*
+ * 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.flink.connector.pulsar.testutils.sink;
+
+import org.apache.flink.connector.testframe.environment.TestEnvironment;
+import org.apache.flink.connector.testframe.external.sink.DataStreamSinkExternalContext;
+import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
+import org.apache.flink.streaming.api.CheckpointingMode;
+
+import org.junit.jupiter.api.Disabled;
+
+/** Pulsar sink don't expose the monitor metrics now. We have to disable this test. */
+public abstract class PulsarSinkTestSuiteBase extends SinkTestSuiteBase<String> {
+
+ @Override
+ @Disabled("Enable this test after FLINK-26027 being merged.")
+ public void testMetrics(
+ TestEnvironment testEnv,
+ DataStreamSinkExternalContext<String> externalContext,
+ CheckpointingMode semantic) {}
+}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/KeyedPulsarPartitionDataWriter.java
similarity index 77%
rename from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/KeyedPulsarPartitionDataWriter.java
index d5f6e11c00a..fa1247b73d8 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/KeyedPulsarPartitionDataWriter.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.common;
+package org.apache.flink.connector.pulsar.testutils.source;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
@@ -25,8 +25,6 @@ import org.apache.pulsar.client.api.Schema;
import java.util.List;
-import static java.util.stream.Collectors.toList;
-
/**
* Source split data writer for writing test data into a Pulsar topic partition. It will write the
* message with two keys.
@@ -36,24 +34,17 @@ public class KeyedPulsarPartitionDataWriter implements ExternalSystemSplitDataWr
private final PulsarRuntimeOperator operator;
private final String fullTopicName;
private final String keyToRead;
- private final String keyToExclude;
public KeyedPulsarPartitionDataWriter(
- PulsarRuntimeOperator operator,
- String fullTopicName,
- String keyToRead,
- String keyToExclude) {
+ PulsarRuntimeOperator operator, String fullTopicName, String keyToRead) {
this.operator = operator;
this.fullTopicName = fullTopicName;
this.keyToRead = keyToRead;
- this.keyToExclude = keyToExclude;
}
@Override
public void writeRecords(List<String> records) {
- List<String> newRecords = records.stream().map(a -> a + keyToRead).collect(toList());
-
- operator.sendMessages(fullTopicName, Schema.STRING, keyToExclude, newRecords);
+ // Send messages with the given key.
operator.sendMessages(fullTopicName, Schema.STRING, keyToRead, records);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
index 35496ccab21..af41d8c3093 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
@@ -72,8 +72,5 @@ public abstract class UnorderedSourceTestSuiteBase<T> extends SourceTestSuiteBas
public void testIdleReader(
TestEnvironment testEnv,
DataStreamSourceExternalContext<T> externalContext,
- CheckpointingMode semantic)
- throws Exception {
- super.testIdleReader(testEnv, externalContext, semantic);
- }
+ CheckpointingMode semantic) {}
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
similarity index 81%
rename from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
index 0cae6e58405..1901787b342 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
@@ -16,16 +16,15 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.source;
+package org.apache.flink.connector.pulsar.testutils.source.cases;
import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.KeyedPulsarPartitionDataWriter;
import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.util.Murmur3_32Hash;
@@ -33,33 +32,23 @@ import org.apache.pulsar.common.util.Murmur3_32Hash;
import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
-import static org.apache.pulsar.client.api.SubscriptionType.Key_Shared;
/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext {
private final String keyToRead;
- private final String keyToExclude;
public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment, Key_Shared);
+ super(environment);
this.keyToRead = randomAlphabetic(8);
-
- // Make sure they have different hash code.
- int readHash = keyHash(keyToRead);
- String randomKey;
- do {
- randomKey = randomAlphabetic(8);
- } while (keyHash(randomKey) == readHash);
- this.keyToExclude = randomKey;
}
@Override
public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
TestingSourceSettings sourceSettings) {
String partitionName = generatePartitionName();
- return new KeyedPulsarPartitionDataWriter(operator, partitionName, keyToRead, keyToExclude);
+ return new KeyedPulsarPartitionDataWriter(operator, partitionName, keyToRead);
}
@Override
@@ -80,6 +69,11 @@ public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext
return "pulsar-key-shared-subscription";
}
+ @Override
+ protected SubscriptionType subscriptionType() {
+ return SubscriptionType.Key_Shared;
+ }
+
// This method is copied from Pulsar for calculating message key hash.
private int keyHash(String key) {
return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
index ca6410fbc43..422c954a4e7 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
@@ -33,18 +33,11 @@ import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNam
public class MultipleTopicConsumingContext extends PulsarSourceTestContext {
private final String topicPrefix = "flink-multiple-topic-" + randomAlphabetic(8) + "-";
- private final SubscriptionType subscriptionType;
private int index = 0;
public MultipleTopicConsumingContext(PulsarTestEnvironment environment) {
- this(environment, SubscriptionType.Exclusive);
- }
-
- public MultipleTopicConsumingContext(
- PulsarTestEnvironment environment, SubscriptionType subscriptionType) {
super(environment);
- this.subscriptionType = subscriptionType;
}
@Override
@@ -64,7 +57,7 @@ public class MultipleTopicConsumingContext extends PulsarSourceTestContext {
@Override
protected SubscriptionType subscriptionType() {
- return subscriptionType;
+ return SubscriptionType.Exclusive;
}
@Override
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SharedSubscriptionContext.java
similarity index 84%
rename from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SharedSubscriptionContext.java
index fe9f07832af..cd649fef040 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SharedSubscriptionContext.java
@@ -16,20 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.source;
+package org.apache.flink.connector.pulsar.testutils.source.cases;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import static org.apache.pulsar.client.api.SubscriptionType.Shared;
-
/** We would consume from test splits by using {@link SubscriptionType#Shared} subscription. */
public class SharedSubscriptionContext extends MultipleTopicConsumingContext {
public SharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment, Shared);
+ super(environment);
}
@Override
@@ -41,4 +38,9 @@ public class SharedSubscriptionContext extends MultipleTopicConsumingContext {
protected String subscriptionName() {
return "pulsar-shared-subscription";
}
+
+ @Override
+ protected SubscriptionType subscriptionType() {
+ return SubscriptionType.Shared;
+ }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/resources/docker/bootstrap.sh b/flink-connectors/flink-connector-pulsar/src/test/resources/docker/bootstrap.sh
deleted file mode 100755
index fa4e2921a75..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/resources/docker/bootstrap.sh
+++ /dev/null
@@ -1,28 +0,0 @@
-#!/usr/bin/env bash
-# ----------------------------------------------------------------------------
-# 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.
-# ----------------------------------------------------------------------------
-
-# Enable the transaction in standalone config.
-sed -i 's/transactionCoordinatorEnabled=false/transactionCoordinatorEnabled=true/g' /pulsar/conf/standalone.conf
-sed -i 's/acknowledgmentAtBatchIndexLevelEnabled=false/acknowledgmentAtBatchIndexLevelEnabled=true/g' /pulsar/conf/standalone.conf
-sed -i 's/systemTopicEnabled=false/systemTopicEnabled=true/g' /pulsar/conf/standalone.conf
-sed -i 's/brokerDeduplicationEnabled=false/brokerDeduplicationEnabled=true/g' /pulsar/conf/standalone.conf
-
-# Start Pulsar standalone without function worker and streaming storage.
-/pulsar/bin/pulsar standalone --no-functions-worker -nss
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
similarity index 64%
copy from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
copy to flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
index 15333a1d538..a19f5931b58 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -19,7 +19,8 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestSuiteBase;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -27,38 +28,30 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
-/**
- * Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
- * subscription.
- */
+/** Pulsar sink E2E test based on connector testing framework. */
@SuppressWarnings("unused")
@Category(value = {FailsOnJava11.class})
-public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
+public class PulsarSinkE2ECase extends PulsarSinkTestSuiteBase {
- // Defines the Semantic.
@TestSemantics
- CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+ CheckpointingMode[] semantics =
+ new CheckpointingMode[] {
+ CheckpointingMode.EXACTLY_ONCE, CheckpointingMode.AT_LEAST_ONCE
+ };
- // Defines TestEnvironment.
+ // Defines TestEnvironment
@TestEnv
- FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 8);
+ FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
// Defines ConnectorExternalSystem.
@TestExternalSystem
PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
- // Defines a set of external context Factories for different test cases.
- @TestContext
- PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
- new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
-
@TestContext
- PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
- new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
+ PulsarTestContextFactory<String, PulsarSinkTestContext> sinkContext =
+ new PulsarTestContextFactory<>(pulsar, PulsarSinkTestContext::new);
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 15333a1d538..89692d1c449 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -20,6 +20,8 @@ package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
+import org.apache.flink.connector.pulsar.testutils.source.cases.KeySharedSubscriptionContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.SharedSubscriptionContext;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -27,8 +29,6 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 14d8cbdd12c..b5ddbb8b976 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -51,11 +51,8 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
return TestUtils.getResource(jarName).toAbsolutePath().toString();
}
- protected static Configuration flinkConfiguration() {
+ private static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the off heap memory of TaskManager to avoid direct buffer memory error in Pulsar
- // e2e tests.
- configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
// Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
[flink] 05/05: [FLINK-29381][Connector/Pulsar] Fixes the split assignment for Key Shared subscription.
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 33b290ec9c4426a5cd08f8749256c55e7e20e93a
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Sep 30 01:40:14 2022 +0800
[FLINK-29381][Connector/Pulsar] Fixes the split assignment for Key Shared subscription.
---
...itAssigner.java => KeySharedSplitAssigner.java} | 65 ++++++++--------------
.../assigner/NonSharedSplitAssigner.java | 33 -----------
.../enumerator/assigner/SplitAssignerBase.java | 33 +++++++++++
.../enumerator/assigner/SplitAssignerFactory.java | 4 +-
.../assigner/NonSharedSplitAssignerTest.java | 2 +-
.../source/cases/KeySharedSubscriptionContext.java | 5 +-
6 files changed, 62 insertions(+), 80 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/KeySharedSplitAssigner.java
similarity index 57%
copy from flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java
copy to flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/KeySharedSplitAssigner.java
index 1b7b4a6f446..9ef43476f71 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/KeySharedSplitAssigner.java
@@ -19,7 +19,6 @@
package org.apache.flink.connector.pulsar.source.enumerator.assigner;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
@@ -32,14 +31,11 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
-/**
- * This assigner is used for {@link SubscriptionType#Failover}, {@link SubscriptionType#Exclusive}
- * and {@link SubscriptionType#Key_Shared} subscriptions.
- */
+/** This assigner is used for {@link SubscriptionType#Key_Shared} subscription. */
@Internal
-class NonSharedSplitAssigner extends SplitAssignerBase {
+public class KeySharedSplitAssigner extends SplitAssignerBase {
- public NonSharedSplitAssigner(
+ public KeySharedSplitAssigner(
StopCursor stopCursor,
boolean enablePartitionDiscovery,
SplitEnumeratorContext<PulsarPartitionSplit> context,
@@ -52,10 +48,15 @@ class NonSharedSplitAssigner extends SplitAssignerBase {
List<TopicPartition> newPartitions = new ArrayList<>();
for (TopicPartition partition : fetchedPartitions) {
+ boolean shouldAssign = false;
if (!appendedPartitions.contains(partition)) {
appendedPartitions.add(partition);
newPartitions.add(partition);
+ shouldAssign = true;
+ }
+ // Reassign the incoming splits when restarting from state.
+ if (shouldAssign || !initialized) {
// Calculate the reader id by the current parallelism.
int readerId = partitionOwner(partition);
PulsarPartitionSplit split = new PulsarPartitionSplit(partition, stopCursor);
@@ -72,41 +73,21 @@ class NonSharedSplitAssigner extends SplitAssignerBase {
@Override
public void addSplitsBack(List<PulsarPartitionSplit> splits, int subtaskId) {
- for (PulsarPartitionSplit split : splits) {
- int readerId = partitionOwner(split.getPartition());
- addSplitToPendingList(readerId, split);
+ if (splits.isEmpty()) {
+ // In case of the task failure. No splits will be put back to the enumerator.
+ for (TopicPartition partition : appendedPartitions) {
+ int readId = partitionOwner(partition);
+ if (readId == subtaskId) {
+ PulsarPartitionSplit split = new PulsarPartitionSplit(partition, stopCursor);
+ addSplitToPendingList(subtaskId, split);
+ }
+ }
+ } else {
+ // Manually put all the splits back to the enumerator.
+ for (PulsarPartitionSplit split : splits) {
+ int readerId = partitionOwner(split.getPartition());
+ addSplitToPendingList(readerId, split);
+ }
}
}
-
- /**
- * Returns the index of the target subtask that a specific partition should be assigned to. It's
- * inspired by the {@code KafkaSourceEnumerator.getSplitOwner()}
- *
- * <p>The resulting distribution of partition has the following contract:
- *
- * <ul>
- * <li>1. Uniformly distributed across subtasks.
- * <li>2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending subtask
- * indices) by using the partition id as the offset from a starting index (i.e., the index
- * of the subtask which partition 0 of the topic will be assigned to, determined using the
- * topic name).
- * </ul>
- *
- * @param partition The Pulsar partition to assign.
- * @return The id of the reader that owns this partition.
- */
- private int partitionOwner(TopicPartition partition) {
- return calculatePartitionOwner(
- partition.getTopic(), partition.getPartitionId(), context.currentParallelism());
- }
-
- @VisibleForTesting
- static int calculatePartitionOwner(String topic, int partitionId, int parallelism) {
- int startIndex = ((topic.hashCode() * 31) & 0x7FFFFFFF) % parallelism;
- /*
- * Here, the assumption is that the id of Pulsar partitions are always ascending starting from
- * 0. Therefore, can be used directly as the offset clockwise from the start index.
- */
- return (startIndex + partitionId) % parallelism;
- }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java
index 1b7b4a6f446..833477c638e 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssigner.java
@@ -19,7 +19,6 @@
package org.apache.flink.connector.pulsar.source.enumerator.assigner;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
@@ -77,36 +76,4 @@ class NonSharedSplitAssigner extends SplitAssignerBase {
addSplitToPendingList(readerId, split);
}
}
-
- /**
- * Returns the index of the target subtask that a specific partition should be assigned to. It's
- * inspired by the {@code KafkaSourceEnumerator.getSplitOwner()}
- *
- * <p>The resulting distribution of partition has the following contract:
- *
- * <ul>
- * <li>1. Uniformly distributed across subtasks.
- * <li>2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending subtask
- * indices) by using the partition id as the offset from a starting index (i.e., the index
- * of the subtask which partition 0 of the topic will be assigned to, determined using the
- * topic name).
- * </ul>
- *
- * @param partition The Pulsar partition to assign.
- * @return The id of the reader that owns this partition.
- */
- private int partitionOwner(TopicPartition partition) {
- return calculatePartitionOwner(
- partition.getTopic(), partition.getPartitionId(), context.currentParallelism());
- }
-
- @VisibleForTesting
- static int calculatePartitionOwner(String topic, int partitionId, int parallelism) {
- int startIndex = ((topic.hashCode() * 31) & 0x7FFFFFFF) % parallelism;
- /*
- * Here, the assumption is that the id of Pulsar partitions are always ascending starting from
- * 0. Therefore, can be used directly as the offset clockwise from the start index.
- */
- return (startIndex + partitionId) % parallelism;
- }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerBase.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerBase.java
index 32af433acbb..733072ca7e4 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerBase.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.pulsar.source.enumerator.assigner;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.connector.source.SplitsAssignment;
import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
@@ -99,4 +100,36 @@ abstract class SplitAssignerBase implements SplitAssigner {
pendingPartitionSplits.computeIfAbsent(readerId, i -> new HashSet<>());
splits.add(split);
}
+
+ /**
+ * Returns the index of the target subtask that a specific partition should be assigned to. It's
+ * inspired by the {@code KafkaSourceEnumerator.getSplitOwner()}
+ *
+ * <p>The resulting distribution of partition has the following contract:
+ *
+ * <ul>
+ * <li>1. Uniformly distributed across subtasks.
+ * <li>2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending subtask
+ * indices) by using the partition id as the offset from a starting index (i.e., the index
+ * of the subtask which partition 0 of the topic will be assigned to, determined using the
+ * topic name).
+ * </ul>
+ *
+ * @param partition The Pulsar partition to assign.
+ * @return The id of the reader that owns this partition.
+ */
+ protected int partitionOwner(TopicPartition partition) {
+ return calculatePartitionOwner(
+ partition.getTopic(), partition.getPartitionId(), context.currentParallelism());
+ }
+
+ @VisibleForTesting
+ static int calculatePartitionOwner(String topic, int partitionId, int parallelism) {
+ int startIndex = ((topic.hashCode() * 31) & 0x7FFFFFFF) % parallelism;
+ /*
+ * Here, the assumption is that the id of Pulsar partitions are always ascending starting from
+ * 0. Therefore, can be used directly as the offset clockwise from the start index.
+ */
+ return (startIndex + partitionId) % parallelism;
+ }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerFactory.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerFactory.java
index 02f8934a6bf..4ade1e561a7 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerFactory.java
@@ -46,12 +46,14 @@ public final class SplitAssignerFactory {
switch (subscriptionType) {
case Failover:
case Exclusive:
- case Key_Shared:
return new NonSharedSplitAssigner(
stopCursor, enablePartitionDiscovery, context, enumState);
case Shared:
return new SharedSplitAssigner(
stopCursor, enablePartitionDiscovery, context, enumState);
+ case Key_Shared:
+ return new KeySharedSplitAssigner(
+ stopCursor, enablePartitionDiscovery, context, enumState);
default:
throw new IllegalArgumentException(
"We don't support this subscription type: " + subscriptionType);
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssignerTest.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssignerTest.java
index 58f8d8fc51c..bc2d669aeb9 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssignerTest.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/NonSharedSplitAssignerTest.java
@@ -36,7 +36,7 @@ import java.util.Optional;
import java.util.Set;
import static java.util.Collections.singletonList;
-import static org.apache.flink.connector.pulsar.source.enumerator.assigner.NonSharedSplitAssigner.calculatePartitionOwner;
+import static org.apache.flink.connector.pulsar.source.enumerator.assigner.SplitAssignerBase.calculatePartitionOwner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
index 1901787b342..7aa689e6106 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/KeySharedSubscriptionContext.java
@@ -29,6 +29,7 @@ import org.apache.flink.connector.testframe.external.source.TestingSourceSetting
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.util.Murmur3_32Hash;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
@@ -40,7 +41,6 @@ public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext
public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
super(environment);
-
this.keyToRead = randomAlphabetic(8);
}
@@ -60,7 +60,6 @@ public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext
protected void setSourceBuilder(PulsarSourceBuilder<String> builder) {
int keyHash = keyHash(keyToRead);
TopicRange range = new TopicRange(keyHash, keyHash);
-
builder.setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
}
@@ -76,6 +75,6 @@ public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext
// This method is copied from Pulsar for calculating message key hash.
private int keyHash(String key) {
- return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
+ return Murmur3_32Hash.getInstance().makeHash(key.getBytes(UTF_8)) % RANGE_SIZE;
}
}
[flink] 01/05: [FLINK-26182][Connector/Pulsar] Drop the embedded Pulsar runtime for in favor of mock Pulsar runtime with local metastore support.
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1ccf832d0913f56b1086ac3d43a76875036a66d2
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Sep 27 18:04:23 2022 +0800
[FLINK-26182][Connector/Pulsar] Drop the embedded Pulsar runtime for in favor of mock Pulsar runtime with local metastore support.
---
.../pulsar/testutils/runtime/PulsarRuntime.java | 17 +-
.../runtime/embedded/PulsarEmbeddedRuntime.java | 184 ---------------------
2 files changed, 2 insertions(+), 199 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
index 92dd94c6657..6beb94a68c4 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.testutils.runtime;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime;
-import org.apache.flink.connector.pulsar.testutils.runtime.embedded.PulsarEmbeddedRuntime;
import org.apache.flink.connector.pulsar.testutils.runtime.mock.PulsarMockRuntime;
import org.testcontainers.containers.GenericContainer;
@@ -50,22 +49,10 @@ public interface PulsarRuntime {
return new PulsarMockRuntime();
}
- /**
- * Create a standalone Pulsar instance in test thread. We would start an embedded zookeeper and
- * bookkeeper. The stream storage for bookkeeper is disabled. The function worker is disabled on
- * Pulsar broker.
- *
- * <p>This runtime would be faster than {@link #container()} and behaves the same as the {@link
- * #container()}.
- */
- static PulsarRuntime embedded() {
- return new PulsarEmbeddedRuntime();
- }
-
/**
* Create a Pulsar instance in docker. We would start a standalone Pulsar in TestContainers.
* This runtime is often used in end-to-end tests. The performance may be a bit of slower than
- * {@link #embedded()}. The stream storage for bookkeeper is disabled. The function worker is
+ * {@link #mock()}. The stream storage for bookkeeper is disabled. The function worker is
* disabled on Pulsar broker.
*/
static PulsarRuntime container() {
@@ -75,7 +62,7 @@ public interface PulsarRuntime {
/**
* Create a Pulsar instance in docker. We would start a standalone Pulsar in TestContainers.
* This runtime is often used in end-to-end tests. The performance may be a bit of slower than
- * {@link #embedded()}. The stream storage for bookkeeper is disabled. The function worker is
+ * {@link #mock()}. The stream storage for bookkeeper is disabled. The function worker is
* disabled on Pulsar broker.
*
* <p>We would link the created Pulsar docker instance with the given flink instance. This would
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/embedded/PulsarEmbeddedRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/embedded/PulsarEmbeddedRuntime.java
deleted file mode 100644
index 2ca9a51f3c5..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/embedded/PulsarEmbeddedRuntime.java
+++ /dev/null
@@ -1,184 +0,0 @@
-/*
- * 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.flink.connector.pulsar.testutils.runtime.embedded;
-
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
-import org.apache.flink.util.FileUtils;
-
-import org.apache.bookkeeper.conf.ServerConfiguration;
-import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
-import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.Optional;
-
-import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeUtils.initializePulsarEnvironment;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.pulsar.broker.ServiceConfigurationUtils.brokerUrl;
-import static org.apache.pulsar.broker.ServiceConfigurationUtils.webServiceUrl;
-
-/** Providing a embedded pulsar server. We use this runtime for transaction related tests. */
-public class PulsarEmbeddedRuntime implements PulsarRuntime {
- private static final Logger LOG = LoggerFactory.getLogger(PulsarEmbeddedRuntime.class);
-
- private static final String CONFIG_FILE_PATH;
-
- static {
- // Find the absolute path for containers/txnStandalone.conf
- ClassLoader classLoader = PulsarEmbeddedRuntime.class.getClassLoader();
- URL resource = classLoader.getResource("containers/txnStandalone.conf");
- File file = new File(checkNotNull(resource).getFile());
- CONFIG_FILE_PATH = file.getAbsolutePath();
- }
-
- private final Path tempDir;
-
- private LocalBookkeeperEnsemble bookkeeper;
- private PulsarService pulsarService;
- private PulsarRuntimeOperator operator;
-
- public PulsarEmbeddedRuntime() {
- this.tempDir = createTempDir();
- }
-
- @Override
- public void startUp() {
- try {
- startBookkeeper();
- startPulsarService();
-
- // Create the operator.
- this.operator = new PulsarRuntimeOperator(serviceUrl(), adminUrl());
- } catch (Exception e) {
- throw new IllegalStateException(e);
- }
- }
-
- @Override
- public void tearDown() {
- try {
- if (operator != null) {
- operator.close();
- }
- if (pulsarService != null) {
- pulsarService.close();
- }
- if (bookkeeper != null) {
- bookkeeper.stop();
- }
- } catch (Exception e) {
- throw new IllegalStateException(e);
- } finally {
- removeTempDir(tempDir);
- }
- }
-
- @Override
- public PulsarRuntimeOperator operator() {
- return checkNotNull(operator, "You should start this embedded Pulsar first.");
- }
-
- private Path createTempDir() {
- try {
- return Files.createTempDirectory("pulsar");
- } catch (IOException e) {
- throw new IllegalStateException(e);
- }
- }
-
- private void removeTempDir(Path tempDir) {
- try {
- FileUtils.deleteDirectory(tempDir.normalize().toFile());
- } catch (IOException e) {
- throw new IllegalStateException(e);
- }
- }
-
- public void startBookkeeper() throws Exception {
- Path zkPath = Paths.get("data", "standalone", "zookeeper");
- Path bkPath = Paths.get("data", "standalone", "bookkeeper");
-
- String zkDir = tempDir.resolve(zkPath).normalize().toString();
- String bkDir = tempDir.resolve(bkPath).normalize().toString();
-
- ServerConfiguration bkServerConf = new ServerConfiguration();
- bkServerConf.loadConf(new File(CONFIG_FILE_PATH).toURI().toURL());
- this.bookkeeper = new LocalBookkeeperEnsemble(1, 0, 0, zkDir, bkDir, true, "127.0.0.1");
-
- // Start Bookkeeper & zookeeper.
- bookkeeper.startStandalone(bkServerConf, false);
- }
-
- private void startPulsarService() throws Exception {
- ServiceConfiguration config;
- try (FileInputStream inputStream = new FileInputStream(CONFIG_FILE_PATH)) {
- config = PulsarConfigurationLoader.create(inputStream, ServiceConfiguration.class);
- } catch (IOException e) {
- throw new IllegalStateException(e);
- }
-
- // Use runtime dynamic ports for broker.
- config.setAdvertisedAddress("127.0.0.1");
- config.setClusterName("standalone");
-
- // Use random port.
- config.setBrokerServicePort(Optional.of(0));
- config.setWebServicePort(Optional.of(0));
-
- // Select available port for bookkeeper and zookeeper.
- int zkPort = getZkPort();
- String zkConnect = "127.0.0.1" + ":" + zkPort;
- config.setZookeeperServers(zkConnect);
- config.setConfigurationStoreServers(zkConnect);
- config.setRunningStandalone(true);
-
- this.pulsarService = new PulsarService(config);
-
- // Start Pulsar Broker.
- pulsarService.start();
-
- // Create sample data environment.
- initializePulsarEnvironment(config, serviceUrl(), adminUrl());
- }
-
- private int getZkPort() {
- return checkNotNull(bookkeeper).getZookeeperPort();
- }
-
- private String serviceUrl() {
- Integer port = pulsarService.getBrokerListenPort().orElseThrow(IllegalStateException::new);
- return brokerUrl("127.0.0.1", port);
- }
-
- private String adminUrl() {
- Integer port = pulsarService.getListenPortHTTP().orElseThrow(IllegalArgumentException::new);
- return webServiceUrl("127.0.0.1", port);
- }
-}
[flink] 03/05: [FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 022df77665e33a00efde71c05b039e816009ecd9
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Sep 27 18:04:24 2022 +0800
[FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
---
.../pulsar/source/enumerator/topic/TopicRange.java | 2 +-
.../pulsar/source/PulsarSourceITCase.java | 6 +-
.../pulsar/source/PulsarUnorderedSourceITCase.java | 60 ++-------
.../pulsar/testutils/PulsarTestCommonUtils.java | 17 ++-
.../pulsar/testutils/PulsarTestContext.java | 45 +++----
.../pulsar/testutils/PulsarTestContextFactory.java | 5 +-
.../pulsar/testutils/PulsarTestEnvironment.java | 2 +-
.../pulsar/testutils/PulsarTestSuiteBase.java | 5 +-
.../cases/MultipleTopicTemplateContext.java | 134 -------------------
.../cases/SharedSubscriptionConsumingContext.java | 58 ---------
.../cases/SingleTopicConsumingContext.java | 132 -------------------
.../{ => source}/PulsarPartitionDataWriter.java | 18 ++-
.../testutils/source/PulsarSourceTestContext.java | 138 ++++++++++++++++++++
.../source/UnorderedSourceTestSuiteBase.java} | 40 ++----
.../cases/MultipleTopicConsumingContext.java | 42 ++++--
.../source/cases/SingleTopicConsumingContext.java | 73 +++++++++++
.../flink-end-to-end-tests-pulsar/pom.xml | 24 +++-
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 5 +-
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 13 +-
.../pulsar/cases/KeySharedSubscriptionContext.java | 144 ---------------------
.../pulsar/cases/SharedSubscriptionContext.java | 116 -----------------
.../FlinkContainerWithPulsarEnvironment.java | 3 +-
.../common/KeyedPulsarPartitionDataWriter.java | 19 +--
.../common/UnorderedSourceTestSuiteBase.java | 86 ------------
.../ExclusiveSubscriptionContext.java | 19 +--
.../FailoverSubscriptionContext.java | 19 +--
.../source/KeySharedSubscriptionContext.java | 87 +++++++++++++
.../pulsar/source/SharedSubscriptionContext.java | 32 ++---
.../source/enumerator/NoOpEnumStateSerializer.java | 2 +-
29 files changed, 466 insertions(+), 880 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
index 5b779224b87..1508b8732a3 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
@@ -68,7 +68,7 @@ public class TopicRange implements Serializable {
return new Range(start, end);
}
- /** Create a topic range which contains the fully hash range. */
+ /** Create a topic range which contains the full hash range. */
public static TopicRange createFullRange() {
return new TopicRange(MIN_RANGE, MAX_RANGE);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
index 54fc4d71dd5..c2c95f255b5 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
@@ -20,9 +20,9 @@ package org.apache.flink.connector.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicConsumingContext;
-import org.apache.flink.connector.pulsar.testutils.cases.SingleTopicConsumingContext;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.SingleTopicConsumingContext;
import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
@@ -41,6 +41,7 @@ import org.junit.experimental.categories.Category;
*/
@Category(value = {FailsOnJava11.class})
class PulsarSourceITCase extends SourceTestSuiteBase<String> {
+
// Defines test environment on Flink MiniCluster
@TestEnv MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
@@ -48,6 +49,7 @@ class PulsarSourceITCase extends SourceTestSuiteBase<String> {
@TestExternalSystem
PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.mock());
+ // This field is preserved, we don't support the semantics in source currently.
@TestSemantics
CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
index 6bad7220886..d18c6c2633b 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
@@ -20,36 +20,28 @@ package org.apache.flink.connector.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.SharedSubscriptionConsumingContext;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
+import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
-import org.apache.flink.connector.testframe.environment.TestEnvironment;
-import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
-import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.testutils.junit.FailsOnJava11;
-import org.apache.flink.util.CloseableIterator;
import org.apache.pulsar.client.api.SubscriptionType;
import org.junit.experimental.categories.Category;
-import org.junit.jupiter.api.Disabled;
-import java.util.List;
-
-import static java.util.concurrent.CompletableFuture.runAsync;
-import static org.apache.flink.connector.testframe.utils.CollectIteratorAssertions.assertUnordered;
-import static org.apache.flink.connector.testframe.utils.ConnectorTestConstants.DEFAULT_COLLECT_DATA_TIMEOUT;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.apache.pulsar.client.api.SubscriptionType.Shared;
/**
* Unit test class for {@link PulsarSource}. Used for {@link SubscriptionType#Shared} subscription.
*/
@Category(value = {FailsOnJava11.class})
-public class PulsarUnorderedSourceITCase extends SourceTestSuiteBase<String> {
+public class PulsarUnorderedSourceITCase extends UnorderedSourceTestSuiteBase<String> {
+
// Defines test environment on Flink MiniCluster
@TestEnv MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
@@ -61,43 +53,7 @@ public class PulsarUnorderedSourceITCase extends SourceTestSuiteBase<String> {
CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
@TestContext
- PulsarTestContextFactory<String, SharedSubscriptionConsumingContext> singleTopic =
- new PulsarTestContextFactory<>(pulsar, SharedSubscriptionConsumingContext::new);
-
- @Override
- protected void checkResultWithSemantic(
- CloseableIterator<String> resultIterator,
- List<List<String>> testData,
- CheckpointingMode semantic,
- Integer limit) {
- Runnable runnable =
- () ->
- assertUnordered(resultIterator)
- .withNumRecordsLimit(getExpectedSize(testData, limit))
- .matchesRecordsFromSource(testData, semantic);
-
- assertThat(runAsync(runnable)).succeedsWithin(DEFAULT_COLLECT_DATA_TIMEOUT);
- }
-
- /**
- * Shared subscription will have multiple readers on same partition, this would make hard to
- * automatically stop like a bounded source.
- */
- private static int getExpectedSize(List<List<String>> testData, Integer limit) {
- if (limit == null) {
- return testData.stream().mapToInt(List::size).sum();
- } else {
- return limit;
- }
- }
-
- @Override
- @Disabled("We don't have any idle readers in Pulsar's shared subscription.")
- public void testIdleReader(
- TestEnvironment testEnv,
- DataStreamSourceExternalContext<String> externalContext,
- CheckpointingMode semantic)
- throws Exception {
- super.testIdleReader(testEnv, externalContext, semantic);
- }
+ PulsarTestContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
+ new PulsarTestContextFactory<>(
+ pulsar, env -> new MultipleTopicConsumingContext(env, Shared));
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
index 87f3976c6a7..96b1ca62f2a 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
@@ -19,10 +19,12 @@
package org.apache.flink.connector.pulsar.testutils;
import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.pulsar.client.api.MessageId;
import org.junit.jupiter.api.extension.ParameterContext;
@@ -33,7 +35,18 @@ import java.util.List;
/** Put static methods that can be used by multiple test classes. */
public class PulsarTestCommonUtils {
- // ------- CreateSplits
+ /** Convert the CheckpointingMode to a connector related DeliveryGuarantee. */
+ public static DeliveryGuarantee toDeliveryGuarantee(CheckpointingMode checkpointingMode) {
+ if (checkpointingMode == CheckpointingMode.AT_LEAST_ONCE) {
+ return DeliveryGuarantee.AT_LEAST_ONCE;
+ } else if (checkpointingMode == CheckpointingMode.EXACTLY_ONCE) {
+ return DeliveryGuarantee.EXACTLY_ONCE;
+ } else {
+ throw new IllegalArgumentException(
+ "Only exactly-once and al-least-once checkpointing mode are supported.");
+ }
+ }
+
/** creates a fullRange() partitionSplit. */
public static PulsarPartitionSplit createPartitionSplit(String topic, int partitionId) {
return createPartitionSplit(topic, partitionId, Boundedness.CONTINUOUS_UNBOUNDED);
@@ -63,8 +76,6 @@ public class PulsarTestCommonUtils {
return splits;
}
- // -------- InvocationContext Utils
-
public static boolean isAssignableFromParameterContext(
Class<?> requiredType, ParameterContext context) {
return requiredType.isAssignableFrom(context.getParameter().getType());
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
index 9eaa24041c7..a4e44441bfd 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
@@ -19,41 +19,30 @@
package org.apache.flink.connector.pulsar.testutils;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
-import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
+import org.apache.flink.connector.testframe.external.ExternalContext;
+
+import org.apache.pulsar.client.api.Schema;
import java.net.URL;
-import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-import java.util.Random;
-
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
-/** Common test context for pulsar based test. */
-public abstract class PulsarTestContext<T> implements DataStreamSourceExternalContext<T> {
+/**
+ * The implementation for Flink connector test tools. Providing the common test case writing
+ * constraint for both source, sink and table API.
+ */
+public abstract class PulsarTestContext<T> implements ExternalContext {
protected final PulsarRuntimeOperator operator;
- protected final List<URL> connectorJarPaths;
+ // The schema used for consuming and producing messages between Pulsar and tests.
+ protected final Schema<T> schema;
- protected PulsarTestContext(PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
+ protected PulsarTestContext(PulsarTestEnvironment environment, Schema<T> schema) {
this.operator = environment.operator();
- this.connectorJarPaths = connectorJarPaths;
- }
-
- // Helper methods for generating data.
-
- protected List<String> generateStringTestData(int splitIndex, long seed) {
- Random random = new Random(seed);
- int recordNum = 300 + random.nextInt(200);
- List<String> records = new ArrayList<>(recordNum);
-
- for (int i = 0; i < recordNum; i++) {
- int length = random.nextInt(40) + 10;
- records.add(splitIndex + "-" + i + "-" + randomAlphanumeric(length));
- }
-
- return records;
+ this.schema = schema;
}
+ /** Implement this method for providing a more friendly test name in IDE. */
protected abstract String displayName();
@Override
@@ -63,6 +52,10 @@ public abstract class PulsarTestContext<T> implements DataStreamSourceExternalCo
@Override
public List<URL> getConnectorJarPaths() {
- return connectorJarPaths;
+ // We don't need any tests jar definition. They are provided in docker-related environments.
+ return Collections.emptyList();
}
+
+ @Override
+ public void close() throws Exception {}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
index c634efffc44..100c2a67d43 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
@@ -18,13 +18,14 @@
package org.apache.flink.connector.pulsar.testutils;
+import org.apache.flink.connector.pulsar.testutils.source.PulsarSourceTestContext;
import org.apache.flink.connector.testframe.external.ExternalContextFactory;
import java.util.function.Function;
/**
- * Factory for creating all the test context that extends {@link PulsarTestContext}. Test context
- * class should have a constructor with {@link PulsarTestEnvironment} arg.
+ * Factory for creating all the test context that extends {@link PulsarSourceTestContext}. Test
+ * context class should have a constructor with single {@link PulsarTestEnvironment} arg.
*/
public class PulsarTestContextFactory<F, T extends PulsarTestContext<F>>
implements ExternalContextFactory<T> {
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
index 0f3fb9e15f9..f921e4b92f1 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
@@ -48,7 +48,7 @@ import java.util.List;
* }</pre>
*
* <p>If you want to use this class in JUnit 5, just simply extends {@link PulsarTestSuiteBase}, all
- * the helper methods in {@code PulsarContainerOperator} is also exposed there.
+ * the helper methods in {@link PulsarRuntimeOperator} is also exposed there.
*/
public class PulsarTestEnvironment
implements BeforeAllCallback, AfterAllCallback, TestResource, TestRule {
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
index b55fdc553b0..6f64ec008b3 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
@@ -30,9 +30,8 @@ import org.junit.jupiter.api.extension.RegisterExtension;
* The base class for the all Pulsar related test sites. It brings up:
*
* <ul>
- * <li>A Zookeeper cluster.
- * <li>Pulsar Broker.
- * <li>A Bookkeeper cluster.
+ * <li>A Pulsar Broker with memory based local metastore.
+ * <li>A standalone Bookkeeper with memory based local metastore.
* </ul>
*
* <p>You just need to write a JUnit 5 test class and extends this suite class. All the helper
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicTemplateContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicTemplateContext.java
deleted file mode 100644
index 3eca9e7f926..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicTemplateContext.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.flink.connector.pulsar.testutils.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
-import org.apache.pulsar.client.api.SubscriptionType;
-
-import java.net.URL;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-
-/**
- * Pulsar external context template that will create multiple topics with only one partitions as
- * source splits.
- */
-public abstract class MultipleTopicTemplateContext extends PulsarTestContext<String> {
-
- private int numTopics = 0;
-
- private final String topicPattern = "pulsar-multiple-topic-[0-9]+-" + randomAlphabetic(8);
-
- private final Map<String, ExternalSystemSplitDataWriter<String>> topicNameToSplitWriters =
- new HashMap<>();
-
- public MultipleTopicTemplateContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public MultipleTopicTemplateContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(serviceUrl())
- .setAdminUrl(adminUrl())
- .setTopicPattern(topicPattern, RegexSubscriptionMode.AllTopics)
- .setSubscriptionType(subscriptionType())
- .setSubscriptionName(subscriptionName());
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- // This is required for SourceTestSuiteBase.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String topicName = topicPattern.replace("[0-9]+", String.valueOf(numTopics));
- operator.createTopic(topicName, 1);
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(operator, partitionName);
-
- topicNameToSplitWriters.put(partitionName, writer);
- numTopics++;
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() throws Exception {
- for (ExternalSystemSplitDataWriter<String> writer : topicNameToSplitWriters.values()) {
- writer.close();
- }
-
- topicNameToSplitWriters.clear();
- }
-
- protected abstract String subscriptionName();
-
- protected abstract SubscriptionType subscriptionType();
-
- protected String serviceUrl() {
- return operator.serviceUrl();
- }
-
- protected String adminUrl() {
- return operator.adminUrl();
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SharedSubscriptionConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SharedSubscriptionConsumingContext.java
deleted file mode 100644
index 8001b6e7300..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SharedSubscriptionConsumingContext.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.flink.connector.pulsar.testutils.cases;
-
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-
-import org.apache.pulsar.client.api.SubscriptionType;
-
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
-
-/**
- * A consuming context with {@link SubscriptionType#Shared}, it's almost the same as {@link
- * MultipleTopicConsumingContext}.
- */
-public class SharedSubscriptionConsumingContext extends MultipleTopicTemplateContext {
-
- public SharedSubscriptionConsumingContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public SharedSubscriptionConsumingContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
- }
-
- @Override
- protected String displayName() {
- return "consuming message with shared subscription";
- }
-
- @Override
- protected String subscriptionName() {
- return "flink-shared-subscription-test";
- }
-
- @Override
- protected SubscriptionType subscriptionType() {
- return SubscriptionType.Shared;
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
deleted file mode 100644
index f5bfa45f32b..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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.flink.connector.pulsar.testutils.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-
-import java.net.URL;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ThreadLocalRandom;
-
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-import static org.apache.pulsar.client.api.SubscriptionType.Exclusive;
-
-/**
- * A Pulsar external context that will create only one topic and use partitions in that topic as
- * source splits.
- */
-public class SingleTopicConsumingContext extends PulsarTestContext<String> {
-
- private static final String TOPIC_NAME_PREFIX = "pulsar-single-topic";
- private final String topicName;
- private final Map<Integer, ExternalSystemSplitDataWriter<String>> partitionToSplitWriter =
- new HashMap<>();
-
- private int numSplits = 0;
-
- public SingleTopicConsumingContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public SingleTopicConsumingContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
- this.topicName =
- TOPIC_NAME_PREFIX + "-" + ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
- }
-
- @Override
- protected String displayName() {
- return "consuming message on single topic";
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(operator.serviceUrl())
- .setAdminUrl(operator.adminUrl())
- .setTopics(topicName)
- .setSubscriptionType(Exclusive)
- .setSubscriptionName("pulsar-single-topic");
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- // This is required for SourceTestSuiteBase.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- if (numSplits == 0) {
- // Create the topic first.
- operator.createTopic(topicName, 1);
- numSplits++;
- } else {
- numSplits++;
- operator.increaseTopicPartitions(topicName, numSplits);
- }
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, numSplits - 1);
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(operator, partitionName);
- partitionToSplitWriter.put(numSplits - 1, writer);
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() throws Exception {
- // Close writer.
- for (ExternalSystemSplitDataWriter<String> writer : partitionToSplitWriter.values()) {
- writer.close();
- }
-
- partitionToSplitWriter.clear();
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarPartitionDataWriter.java
similarity index 69%
rename from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarPartitionDataWriter.java
index da48e26cd25..1ceb2922bc3 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarPartitionDataWriter.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils;
+package org.apache.flink.connector.pulsar.testutils.source;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
@@ -25,20 +25,26 @@ import org.apache.pulsar.client.api.Schema;
import java.util.List;
-/** Source split data writer for writing test data into a Pulsar topic partition. */
-public class PulsarPartitionDataWriter implements ExternalSystemSplitDataWriter<String> {
+/**
+ * Source split data writer for writing test data into a Pulsar topic partition. This writer doesn't
+ * need to be closed.
+ */
+public class PulsarPartitionDataWriter<T> implements ExternalSystemSplitDataWriter<T> {
private final PulsarRuntimeOperator operator;
private final String fullTopicName;
+ private final Schema<T> schema;
- public PulsarPartitionDataWriter(PulsarRuntimeOperator operator, String fullTopicName) {
+ public PulsarPartitionDataWriter(
+ PulsarRuntimeOperator operator, String fullTopicName, Schema<T> schema) {
this.operator = operator;
this.fullTopicName = fullTopicName;
+ this.schema = schema;
}
@Override
- public void writeRecords(List<String> records) {
- operator.sendMessages(fullTopicName, Schema.STRING, records);
+ public void writeRecords(List<T> records) {
+ operator.sendMessages(fullTopicName, schema, records);
}
@Override
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
new file mode 100644
index 00000000000..8089f8c58e7
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
@@ -0,0 +1,138 @@
+/*
+ * 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.flink.connector.pulsar.testutils.source;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.pulsar.source.PulsarSource;
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
+
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
+import static org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics;
+
+/**
+ * Common source test context for pulsar based test. We use the string text as the basic send
+ * content.
+ */
+public abstract class PulsarSourceTestContext extends PulsarTestContext<String>
+ implements DataStreamSourceExternalContext<String> {
+
+ private static final long DISCOVERY_INTERVAL = 1000L;
+ private static final int BATCH_DATA_SIZE = 300;
+
+ protected PulsarSourceTestContext(PulsarTestEnvironment environment) {
+ super(environment, Schema.STRING);
+ }
+
+ @Override
+ public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
+ PulsarSourceBuilder<String> builder =
+ PulsarSource.builder()
+ .setDeserializationSchema(pulsarSchema(schema))
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
+ .setTopicPattern(topicPattern(), AllTopics)
+ .setSubscriptionType(subscriptionType())
+ .setSubscriptionName(subscriptionName())
+ .setConfig(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, DISCOVERY_INTERVAL);
+
+ // Set extra configuration for source builder.
+ setSourceBuilder(builder);
+
+ if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
+ // Using the latest stop cursor for making sure the source could be stopped.
+ // This is required for SourceTestSuiteBase.
+ builder.setBoundedStopCursor(StopCursor.latest());
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
+ TestingSourceSettings sourceSettings) {
+ String partitionName = generatePartitionName();
+ return new PulsarPartitionDataWriter<>(operator, partitionName, schema);
+ }
+
+ @Override
+ public List<String> generateTestData(
+ TestingSourceSettings sourceSettings, int splitIndex, long seed) {
+ Random random = new Random(seed);
+ return IntStream.range(0, BATCH_DATA_SIZE)
+ .boxed()
+ .map(
+ index -> {
+ int length = random.nextInt(20) + 1;
+ return "split:"
+ + splitIndex
+ + "-index:"
+ + index
+ + "-content:"
+ + randomAlphanumeric(length);
+ })
+ .collect(toList());
+ }
+
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return Types.STRING;
+ }
+
+ /** Override this method for creating builder. */
+ protected void setSourceBuilder(PulsarSourceBuilder<String> builder) {
+ // Nothing to do by default.
+ }
+
+ /**
+ * The topic pattern which is used in Pulsar topic auto discovery. It was discovered every
+ * {@link #DISCOVERY_INTERVAL} ms;
+ */
+ protected abstract String topicPattern();
+
+ /** The subscription name used in Pulsar consumer. */
+ protected abstract String subscriptionName();
+
+ /** The subscription type used in Pulsar consumer. */
+ protected abstract SubscriptionType subscriptionType();
+
+ /**
+ * Dynamic generate a partition related topic in Pulsar. This topic should be pre-created in
+ * Pulsar. Everytime we call this method, we may get a new partition name.
+ */
+ protected abstract String generatePartitionName();
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
similarity index 58%
copy from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
copy to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
index 6bad7220886..35496ccab21 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/UnorderedSourceTestSuiteBase.java
@@ -16,26 +16,15 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.source;
+package org.apache.flink.connector.pulsar.testutils.source;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.SharedSubscriptionConsumingContext;
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
-import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
import org.apache.flink.connector.testframe.environment.TestEnvironment;
import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
-import org.apache.flink.connector.testframe.junit.annotations.TestContext;
-import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
-import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
-import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.testutils.junit.FailsOnJava11;
import org.apache.flink.util.CloseableIterator;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.junit.experimental.categories.Category;
import org.junit.jupiter.api.Disabled;
import java.util.List;
@@ -46,28 +35,15 @@ import static org.apache.flink.connector.testframe.utils.ConnectorTestConstants.
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
/**
- * Unit test class for {@link PulsarSource}. Used for {@link SubscriptionType#Shared} subscription.
+ * Pulsar with {@link SubscriptionType#Key_Shared} and {@link SubscriptionType#Shared} consumes the
+ * message out of order. So we have to override the default connector test tool.
*/
-@Category(value = {FailsOnJava11.class})
-public class PulsarUnorderedSourceITCase extends SourceTestSuiteBase<String> {
- // Defines test environment on Flink MiniCluster
- @TestEnv MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
-
- // Defines pulsar running environment
- @TestExternalSystem
- PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.mock());
-
- @TestSemantics
- CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
-
- @TestContext
- PulsarTestContextFactory<String, SharedSubscriptionConsumingContext> singleTopic =
- new PulsarTestContextFactory<>(pulsar, SharedSubscriptionConsumingContext::new);
+public abstract class UnorderedSourceTestSuiteBase<T> extends SourceTestSuiteBase<T> {
@Override
protected void checkResultWithSemantic(
- CloseableIterator<String> resultIterator,
- List<List<String>> testData,
+ CloseableIterator<T> resultIterator,
+ List<List<T>> testData,
CheckpointingMode semantic,
Integer limit) {
Runnable runnable =
@@ -83,7 +59,7 @@ public class PulsarUnorderedSourceITCase extends SourceTestSuiteBase<String> {
* Shared subscription will have multiple readers on same partition, this would make hard to
* automatically stop like a bounded source.
*/
- private static int getExpectedSize(List<List<String>> testData, Integer limit) {
+ private static <T> int getExpectedSize(List<List<T>> testData, Integer limit) {
if (limit == null) {
return testData.stream().mapToInt(List::size).sum();
} else {
@@ -95,7 +71,7 @@ public class PulsarUnorderedSourceITCase extends SourceTestSuiteBase<String> {
@Disabled("We don't have any idle readers in Pulsar's shared subscription.")
public void testIdleReader(
TestEnvironment testEnv,
- DataStreamSourceExternalContext<String> externalContext,
+ DataStreamSourceExternalContext<T> externalContext,
CheckpointingMode semantic)
throws Exception {
super.testIdleReader(testEnv, externalContext, semantic);
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
similarity index 52%
copy from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
copy to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
index 57027f33e1b..ca6410fbc43 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
@@ -16,43 +16,63 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils.cases;
+package org.apache.flink.connector.pulsar.testutils.source.cases;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.source.PulsarSourceTestContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
/**
* Pulsar external context that will create multiple topics with only one partitions as source
* splits.
*/
-public class MultipleTopicConsumingContext extends MultipleTopicTemplateContext {
+public class MultipleTopicConsumingContext extends PulsarSourceTestContext {
+
+ private final String topicPrefix = "flink-multiple-topic-" + randomAlphabetic(8) + "-";
+ private final SubscriptionType subscriptionType;
+
+ private int index = 0;
public MultipleTopicConsumingContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
+ this(environment, SubscriptionType.Exclusive);
}
public MultipleTopicConsumingContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ PulsarTestEnvironment environment, SubscriptionType subscriptionType) {
+ super(environment);
+ this.subscriptionType = subscriptionType;
}
@Override
protected String displayName() {
- return "consuming message on multiple topic";
+ return "consume message on multiple topic";
+ }
+
+ @Override
+ protected String topicPattern() {
+ return topicPrefix + ".+";
}
@Override
protected String subscriptionName() {
- return "flink-pulsar-multiple-topic-test";
+ return "flink-multiple-topic-test";
}
@Override
protected SubscriptionType subscriptionType() {
- return SubscriptionType.Exclusive;
+ return subscriptionType;
+ }
+
+ @Override
+ protected String generatePartitionName() {
+ String topic = topicPrefix + index;
+ operator.createTopic(topic, 1);
+ index++;
+
+ return topicNameWithPartition(topic, 0);
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
new file mode 100644
index 00000000000..df5166926e0
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.connector.pulsar.testutils.source.cases;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.source.PulsarSourceTestContext;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
+
+/**
+ * A Pulsar external context that will create only one topic and use partitions in that topic as
+ * source splits.
+ */
+public class SingleTopicConsumingContext extends PulsarSourceTestContext {
+
+ private final String topicName = "pulsar-single-topic-" + randomAlphanumeric(8);
+
+ private int index = 0;
+
+ public SingleTopicConsumingContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consume message on single topic";
+ }
+
+ @Override
+ protected String topicPattern() {
+ return topicName + ".+";
+ }
+
+ @Override
+ protected String subscriptionName() {
+ return "pulsar-single-topic-test";
+ }
+
+ @Override
+ protected SubscriptionType subscriptionType() {
+ return SubscriptionType.Exclusive;
+ }
+
+ @Override
+ protected String generatePartitionName() {
+ if (index == 0) {
+ operator.createTopic(topicName, index + 1);
+ } else {
+ operator.increaseTopicPartitions(topicName, index + 1);
+ }
+
+ return topicNameWithPartition(topicName, index++);
+ }
+}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/pom.xml
index 898c4cbcc2a..8a1f0d962cb 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/pom.xml
@@ -33,6 +33,7 @@ under the License.
<properties>
<pulsar.version>2.10.0</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
+ <jaxb-api.version>2.3.1</jaxb-api.version>
</properties>
<dependencies>
@@ -41,6 +42,17 @@ under the License.
<artifactId>flink-end-to-end-tests-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
@@ -70,9 +82,6 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <skip>true</skip>
- </configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -88,6 +97,15 @@ under the License.
</executions>
<configuration>
<artifactItems>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <destFileName>flink-connector-testing.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies
+ </outputDirectory>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 234c1a01cd6..ea6a982044b 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -25,10 +25,10 @@ import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
+import org.apache.flink.tests.util.pulsar.source.ExclusiveSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.source.FailoverSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
@@ -37,6 +37,7 @@ import org.junit.experimental.categories.Category;
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
+@SuppressWarnings("unused")
@Category(value = {FailsOnJava11.class})
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 50390486dd6..15333a1d538 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -19,21 +19,26 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.tests.util.pulsar.cases.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
+import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
+import org.apache.flink.testutils.junit.FailsOnJava11;
+
+import org.junit.experimental.categories.Category;
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
*/
+@SuppressWarnings("unused")
+@Category(value = {FailsOnJava11.class})
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
// Defines the Semantic.
@@ -49,12 +54,10 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
// Defines a set of external context Factories for different test cases.
- @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
- @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
deleted file mode 100644
index 5ad369bcf03..00000000000
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * 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.flink.tests.util.pulsar.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
-
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
-import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import static java.util.Collections.singletonList;
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-
-/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
-public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
-
- private int index = 0;
-
- private final List<KeyedPulsarPartitionDataWriter> writers = new ArrayList<>();
-
- // Message keys.
- private final String key1;
- private final String key2;
-
- public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public KeySharedSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
-
- // Init message keys.
- this.key1 = randomAlphabetic(8);
- String newKey2;
- do {
- newKey2 = randomAlphabetic(8);
- } while (keyHash(key1) == keyHash(newKey2));
- this.key2 = newKey2;
- }
-
- @Override
- protected String displayName() {
- return "consuming message by Key_Shared";
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- int keyHash = keyHash(key1);
- TopicRange range = new TopicRange(keyHash, keyHash);
-
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(operator.serviceUrl())
- .setAdminUrl(operator.adminUrl())
- .setTopicPattern(
- "pulsar-[0-9]+-key-shared", RegexSubscriptionMode.AllTopics)
- .setSubscriptionType(SubscriptionType.Key_Shared)
- .setSubscriptionName("pulsar-key-shared")
- .setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String topicName = "pulsar-" + index + "-key-shared";
- operator.createTopic(topicName, 1);
- index++;
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
- KeyedPulsarPartitionDataWriter writer =
- new KeyedPulsarPartitionDataWriter(operator, partitionName, key1, key2);
- writers.add(writer);
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() {
- for (KeyedPulsarPartitionDataWriter writer : writers) {
- writer.close();
- }
- writers.clear();
- }
-
- private int keyHash(String key) {
- return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
- }
-}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
deleted file mode 100644
index 1a2db6694d1..00000000000
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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.flink.tests.util.pulsar.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
-import org.apache.pulsar.client.api.SubscriptionType;
-
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-
-/** We would consuming from test splits by using {@link SubscriptionType#Shared} subscription. */
-public class SharedSubscriptionContext extends PulsarTestContext<String> {
-
- private int index = 0;
-
- private final List<PulsarPartitionDataWriter> writers = new ArrayList<>();
-
- public SharedSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public SharedSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
- }
-
- @Override
- protected String displayName() {
- return "consuming message by Shared";
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(operator.serviceUrl())
- .setAdminUrl(operator.adminUrl())
- .setTopicPattern("pulsar-[0-9]+-shared", RegexSubscriptionMode.AllTopics)
- .setSubscriptionType(SubscriptionType.Shared)
- .setSubscriptionName("pulsar-shared");
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String topicName = "pulsar-" + index + "-shared";
- operator.createTopic(topicName, 1);
- index++;
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(operator, partitionName);
- writers.add(writer);
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() {
- for (PulsarPartitionDataWriter writer : writers) {
- writer.close();
- }
- writers.clear();
- }
-}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 9a1c6dd6f0d..14d8cbdd12c 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -43,7 +43,8 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
resourcePath("bcutil-jdk15on.jar"),
resourcePath("bcprov-ext-jdk15on.jar"),
resourcePath("jaxb-api.jar"),
- resourcePath("jul-to-slf4j.jar"));
+ resourcePath("jul-to-slf4j.jar"),
+ resourcePath("flink-connector-testing.jar"));
}
private static String resourcePath(String jarName) {
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
index e431e4c89d0..d5f6e11c00a 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
@@ -35,23 +35,26 @@ public class KeyedPulsarPartitionDataWriter implements ExternalSystemSplitDataWr
private final PulsarRuntimeOperator operator;
private final String fullTopicName;
- private final String key1;
- private final String key2;
+ private final String keyToRead;
+ private final String keyToExclude;
public KeyedPulsarPartitionDataWriter(
- PulsarRuntimeOperator operator, String fullTopicName, String key1, String key2) {
+ PulsarRuntimeOperator operator,
+ String fullTopicName,
+ String keyToRead,
+ String keyToExclude) {
this.operator = operator;
this.fullTopicName = fullTopicName;
- this.key1 = key1;
- this.key2 = key2;
+ this.keyToRead = keyToRead;
+ this.keyToExclude = keyToExclude;
}
@Override
public void writeRecords(List<String> records) {
- operator.sendMessages(fullTopicName, Schema.STRING, key1, records);
+ List<String> newRecords = records.stream().map(a -> a + keyToRead).collect(toList());
- List<String> newRecords = records.stream().map(a -> a + key1).collect(toList());
- operator.sendMessages(fullTopicName, Schema.STRING, key2, newRecords);
+ operator.sendMessages(fullTopicName, Schema.STRING, keyToExclude, newRecords);
+ operator.sendMessages(fullTopicName, Schema.STRING, keyToRead, records);
}
@Override
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
deleted file mode 100644
index 01527ea5482..00000000000
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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.flink.tests.util.pulsar.common;
-
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.testframe.environment.TestEnvironment;
-import org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.connector.testframe.junit.extensions.ConnectorTestingExtension;
-import org.apache.flink.connector.testframe.junit.extensions.TestCaseInvocationContextProvider;
-import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.util.TestLoggerExtension;
-
-import org.junit.jupiter.api.DisplayName;
-import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-import java.util.List;
-import java.util.concurrent.ThreadLocalRandom;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-
-/** A source test template for testing the messages which could be consumed in a unordered way. */
-@ExtendWith({
- ConnectorTestingExtension.class,
- TestLoggerExtension.class,
- TestCaseInvocationContextProvider.class
-})
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public abstract class UnorderedSourceTestSuiteBase<T> {
-
- @TestTemplate
- @DisplayName("Test source with one split and four consumers")
- public void testOneSplitWithMultipleConsumers(
- TestEnvironment testEnv, DataStreamSourceExternalContext<T> externalContext)
- throws Exception {
- TestingSourceSettings sourceSettings =
- TestingSourceSettings.builder()
- .setBoundedness(Boundedness.BOUNDED)
- .setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
- .build();
- TestEnvironmentSettings envOptions =
- TestEnvironmentSettings.builder()
- .setConnectorJarPaths(externalContext.getConnectorJarPaths())
- .build();
- List<T> testData =
- externalContext.generateTestData(
- sourceSettings, 0, ThreadLocalRandom.current().nextLong());
- ExternalSystemSplitDataWriter<T> writer =
- externalContext.createSourceSplitDataWriter(sourceSettings);
- writer.writeRecords(testData);
-
- Source<T, ?, ?> source = externalContext.createSource(sourceSettings);
- StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment(envOptions);
- List<T> results =
- execEnv.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar source")
- .setParallelism(4)
- .executeAndCollect(
- "Source single split with four readers.", testData.size());
-
- assertThat(results, containsInAnyOrder(testData.toArray()));
- }
-}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
similarity index 71%
rename from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
rename to flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
index 6fea0c9c9e5..4906ad6cc7e 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
@@ -16,32 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
-
/** We would consume from test splits by using {@link SubscriptionType#Exclusive} subscription. */
-public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
+public class ExclusiveSubscriptionContext extends MultipleTopicConsumingContext {
public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public ExclusiveSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ super(environment);
}
@Override
protected String displayName() {
- return "consuming message by Exclusive";
+ return "consume message by Exclusive";
}
@Override
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
similarity index 71%
rename from flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
rename to flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
index c47348861c1..3134db4bdda 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
@@ -16,32 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
-
/** We would consume from test splits by using {@link SubscriptionType#Failover} subscription. */
-public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
+public class FailoverSubscriptionContext extends MultipleTopicConsumingContext {
public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public FailoverSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ super(environment);
}
@Override
protected String displayName() {
- return "consuming message by Failover";
+ return "consume message by Failover";
}
@Override
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
new file mode 100644
index 00000000000..0cae6e58405
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.tests.util.pulsar.source;
+
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
+import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
+
+import static java.util.Collections.singletonList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
+import static org.apache.pulsar.client.api.SubscriptionType.Key_Shared;
+
+/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
+public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext {
+
+ private final String keyToRead;
+ private final String keyToExclude;
+
+ public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment, Key_Shared);
+
+ this.keyToRead = randomAlphabetic(8);
+
+ // Make sure they have different hash code.
+ int readHash = keyHash(keyToRead);
+ String randomKey;
+ do {
+ randomKey = randomAlphabetic(8);
+ } while (keyHash(randomKey) == readHash);
+ this.keyToExclude = randomKey;
+ }
+
+ @Override
+ public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
+ TestingSourceSettings sourceSettings) {
+ String partitionName = generatePartitionName();
+ return new KeyedPulsarPartitionDataWriter(operator, partitionName, keyToRead, keyToExclude);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consume message by Key_Shared";
+ }
+
+ @Override
+ protected void setSourceBuilder(PulsarSourceBuilder<String> builder) {
+ int keyHash = keyHash(keyToRead);
+ TopicRange range = new TopicRange(keyHash, keyHash);
+
+ builder.setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
+ }
+
+ @Override
+ protected String subscriptionName() {
+ return "pulsar-key-shared-subscription";
+ }
+
+ // This method is copied from Pulsar for calculating message key hash.
+ private int keyHash(String key) {
+ return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
similarity index 55%
rename from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
rename to flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
index 57027f33e1b..fe9f07832af 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
@@ -16,43 +16,29 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
+import static org.apache.pulsar.client.api.SubscriptionType.Shared;
-/**
- * Pulsar external context that will create multiple topics with only one partitions as source
- * splits.
- */
-public class MultipleTopicConsumingContext extends MultipleTopicTemplateContext {
-
- public MultipleTopicConsumingContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
+/** We would consume from test splits by using {@link SubscriptionType#Shared} subscription. */
+public class SharedSubscriptionContext extends MultipleTopicConsumingContext {
- public MultipleTopicConsumingContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ public SharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment, Shared);
}
@Override
protected String displayName() {
- return "consuming message on multiple topic";
+ return "consume message by Shared";
}
@Override
protected String subscriptionName() {
- return "flink-pulsar-multiple-topic-test";
- }
-
- @Override
- protected SubscriptionType subscriptionType() {
- return SubscriptionType.Exclusive;
+ return "pulsar-shared-subscription";
}
}
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/source/enumerator/NoOpEnumStateSerializer.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/source/enumerator/NoOpEnumStateSerializer.java
index 7be0e8d2d88..0628e632f19 100644
--- a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/source/enumerator/NoOpEnumStateSerializer.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/source/enumerator/NoOpEnumStateSerializer.java
@@ -22,7 +22,7 @@ import org.apache.flink.core.io.SimpleVersionedSerializer;
import java.io.IOException;
-/** Mock enumerator state seializer. */
+/** Mock enumerator state serializer. */
public class NoOpEnumStateSerializer implements SimpleVersionedSerializer<NoOpEnumState> {
@Override
public int getVersion() {