This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit d6e2e06964c4cf40a5e58d4bc2f976a65b4a566b
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Nov 8 14:53:22 2022 +0800
[FLINK-28083][Connector/Pulsar] PulsarSource work with object-reusing
DeserializationSchema. (#21205)
---
.../source/reader/PulsarSourceReaderFactory.java | 23 +++----
.../source/reader/emitter/PulsarRecordEmitter.java | 63 ++++++++++++++++--
.../reader/fetcher/PulsarFetcherManagerBase.java | 27 ++++----
.../fetcher/PulsarOrderedFetcherManager.java | 18 +++---
.../fetcher/PulsarUnorderedFetcherManager.java | 14 ++--
.../source/reader/message/PulsarMessage.java | 74 ----------------------
.../reader/message/PulsarMessageCollector.java | 60 ------------------
.../reader/source/PulsarOrderedSourceReader.java | 15 +++--
.../reader/source/PulsarSourceReaderBase.java | 18 ++----
.../reader/source/PulsarUnorderedSourceReader.java | 13 ++--
.../split/PulsarOrderedPartitionSplitReader.java | 13 +---
.../split/PulsarPartitionSplitReaderBase.java | 33 +++-------
.../split/PulsarUnorderedPartitionSplitReader.java | 11 +---
.../source/enumerator/cursor/StopCursorTest.java | 17 ++---
.../PulsarOrderedPartitionSplitReaderTest.java | 8 +--
.../split/PulsarPartitionSplitReaderTestBase.java | 74 ++++++++++------------
16 files changed, 175 insertions(+), 306 deletions(-)
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
index 820888a..2e83ab5 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
@@ -25,7 +25,7 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
+import
org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarOrderedSourceReader;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarUnorderedSourceReader;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarOrderedPartitionSplitReader;
@@ -33,6 +33,7 @@ import
org.apache.flink.connector.pulsar.source.reader.split.PulsarUnorderedPart
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
@@ -70,25 +71,25 @@ public final class PulsarSourceReaderFactory {
// Create a message queue with the predefined source option.
int queueCapacity = sourceConfiguration.getMessageQueueCapacity();
- FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<OUT>>>
elementsQueue =
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue =
new FutureCompletingBlockingQueue<>(queueCapacity);
+ PulsarRecordEmitter<OUT> recordEmitter = new
PulsarRecordEmitter<>(deserializationSchema);
+
// Create different pulsar source reader by subscription type.
SubscriptionType subscriptionType =
sourceConfiguration.getSubscriptionType();
if (subscriptionType == SubscriptionType.Failover
|| subscriptionType == SubscriptionType.Exclusive) {
// Create an ordered split reader supplier.
- Supplier<PulsarOrderedPartitionSplitReader<OUT>>
splitReaderSupplier =
+ Supplier<PulsarOrderedPartitionSplitReader> splitReaderSupplier =
() ->
- new PulsarOrderedPartitionSplitReader<>(
- pulsarClient,
- pulsarAdmin,
- sourceConfiguration,
- deserializationSchema);
+ new PulsarOrderedPartitionSplitReader(
+ pulsarClient, pulsarAdmin,
sourceConfiguration);
return new PulsarOrderedSourceReader<>(
elementsQueue,
splitReaderSupplier,
+ recordEmitter,
readerContext,
sourceConfiguration,
pulsarClient,
@@ -102,18 +103,18 @@ public final class PulsarSourceReaderFactory {
throw new IllegalStateException("Transaction is required but
didn't enabled");
}
- Supplier<PulsarUnorderedPartitionSplitReader<OUT>>
splitReaderSupplier =
+ Supplier<PulsarUnorderedPartitionSplitReader> splitReaderSupplier =
() ->
- new PulsarUnorderedPartitionSplitReader<>(
+ new PulsarUnorderedPartitionSplitReader(
pulsarClient,
pulsarAdmin,
sourceConfiguration,
- deserializationSchema,
coordinatorClient);
return new PulsarUnorderedSourceReader<>(
elementsQueue,
splitReaderSupplier,
+ recordEmitter,
readerContext,
sourceConfiguration,
pulsarClient,
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/emitter/PulsarRecordEmitter.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/emitter/PulsarRecordEmitter.java
index f6607f0..5cac7de 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/emitter/PulsarRecordEmitter.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/emitter/PulsarRecordEmitter.java
@@ -20,10 +20,13 @@ package
org.apache.flink.connector.pulsar.source.reader.emitter;
import org.apache.flink.api.connector.source.SourceOutput;
import org.apache.flink.connector.base.source.reader.RecordEmitter;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
+import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarOrderedSourceReader;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarUnorderedSourceReader;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
/**
* The {@link RecordEmitter} implementation for both {@link
PulsarOrderedSourceReader} and {@link
@@ -31,15 +34,61 @@ import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
* emitter.
*/
public class PulsarRecordEmitter<T>
- implements RecordEmitter<PulsarMessage<T>, T,
PulsarPartitionSplitState> {
+ implements RecordEmitter<Message<byte[]>, T,
PulsarPartitionSplitState> {
+
+ private final PulsarDeserializationSchema<T> deserializationSchema;
+ private final SourceOutputWrapper<T> sourceOutputWrapper;
+
+ public PulsarRecordEmitter(PulsarDeserializationSchema<T>
deserializationSchema) {
+ this.deserializationSchema = deserializationSchema;
+ this.sourceOutputWrapper = new SourceOutputWrapper<>();
+ }
@Override
public void emitRecord(
- PulsarMessage<T> element, SourceOutput<T> output,
PulsarPartitionSplitState splitState)
+ Message<byte[]> element, SourceOutput<T> output,
PulsarPartitionSplitState splitState)
throws Exception {
- // Sink the record to source output.
- output.collect(element.getValue(), element.getEventTime());
- // Update the split state.
- splitState.setLatestConsumedId(element.getId());
+ // Update the source output.
+ sourceOutputWrapper.setSourceOutput(output);
+ sourceOutputWrapper.setTimestamp(element);
+
+ // Deserialize the message and since it to output.
+ deserializationSchema.deserialize(element, sourceOutputWrapper);
+ splitState.setLatestConsumedId(element.getMessageId());
+
+ // Release the messages if we use message pool in Pulsar.
+ element.release();
+ }
+
+ private static class SourceOutputWrapper<T> implements Collector<T> {
+
+ private SourceOutput<T> sourceOutput;
+ private long timestamp;
+
+ @Override
+ public void collect(T record) {
+ if (timestamp > 0) {
+ sourceOutput.collect(record, timestamp);
+ } else {
+ sourceOutput.collect(record);
+ }
+ }
+
+ @Override
+ public void close() {
+ // Nothing to do here.
+ }
+
+ private void setSourceOutput(SourceOutput<T> sourceOutput) {
+ this.sourceOutput = sourceOutput;
+ }
+
+ /**
+ * Get the event timestamp from Pulsar. Zero means there is no event
time. See {@link
+ * Message#getEventTime()} to get the reason why it returns zero.
+ */
+ private void setTimestamp(Message<?> message) {
+ this.timestamp = message.getEventTime();
+ }
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
index 1622eee..af9bdd4 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
@@ -26,9 +26,10 @@ import
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
import
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.pulsar.client.api.Message;
+
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -36,14 +37,10 @@ import java.util.function.Supplier;
import static java.util.Collections.singletonList;
-/**
- * Common fetcher manager abstraction for both ordered & unordered message.
- *
- * @param <T> The decoded message type for flink.
- */
+/** Common fetcher manager abstraction for both ordered & unordered message. */
@Internal
-public abstract class PulsarFetcherManagerBase<T>
- extends SplitFetcherManager<PulsarMessage<T>, PulsarPartitionSplit> {
+public abstract class PulsarFetcherManagerBase
+ extends SplitFetcherManager<Message<byte[]>, PulsarPartitionSplit> {
private final Map<String, Integer> splitFetcherMapping = new HashMap<>();
private final Map<Integer, Boolean> fetcherStatus = new HashMap<>();
@@ -57,8 +54,8 @@ public abstract class PulsarFetcherManagerBase<T>
* @param splitReaderSupplier The factory for the split reader that
connects to the source
*/
protected PulsarFetcherManagerBase(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<T>>>
elementsQueue,
- Supplier<SplitReader<PulsarMessage<T>, PulsarPartitionSplit>>
splitReaderSupplier,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>>
splitReaderSupplier,
Configuration configuration) {
super(elementsQueue, splitReaderSupplier, configuration);
}
@@ -70,7 +67,7 @@ public abstract class PulsarFetcherManagerBase<T>
@Override
public void addSplits(List<PulsarPartitionSplit> splitsToAdd) {
for (PulsarPartitionSplit split : splitsToAdd) {
- SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> fetcher =
+ SplitFetcher<Message<byte[]>, PulsarPartitionSplit> fetcher =
getOrCreateFetcher(split.splitId());
fetcher.addSplits(singletonList(split));
// This method could be executed multiple times.
@@ -79,7 +76,7 @@ public abstract class PulsarFetcherManagerBase<T>
}
@Override
- protected void startFetcher(SplitFetcher<PulsarMessage<T>,
PulsarPartitionSplit> fetcher) {
+ protected void startFetcher(SplitFetcher<Message<byte[]>,
PulsarPartitionSplit> fetcher) {
if (fetcherStatus.get(fetcher.fetcherId()) != Boolean.TRUE) {
fetcherStatus.put(fetcher.fetcherId(), true);
super.startFetcher(fetcher);
@@ -91,16 +88,16 @@ public abstract class PulsarFetcherManagerBase<T>
Integer fetchId = splitFetcherMapping.remove(splitId);
if (fetchId != null) {
fetcherStatus.remove(fetchId);
- SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> fetcher =
fetchers.remove(fetchId);
+ SplitFetcher<Message<byte[]>, PulsarPartitionSplit> fetcher =
fetchers.remove(fetchId);
if (fetcher != null) {
fetcher.shutdown();
}
}
}
- protected SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit>
getOrCreateFetcher(
+ protected SplitFetcher<Message<byte[]>, PulsarPartitionSplit>
getOrCreateFetcher(
String splitId) {
- SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> fetcher;
+ SplitFetcher<Message<byte[]>, PulsarPartitionSplit> fetcher;
Integer fetcherId = splitFetcherMapping.get(splitId);
if (fetcherId == null) {
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java
index 103dc62..0bc7c98 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java
@@ -25,11 +25,11 @@ import
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarOrderedPartitionSplitReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,16 +41,14 @@ import java.util.function.Supplier;
* Pulsar's FetcherManager implementation for ordered consuming. This class is
needed to help
* acknowledge the message to Pulsar using the {@link Consumer} inside the
{@link
* PulsarOrderedPartitionSplitReader}.
- *
- * @param <T> The message type for pulsar decoded message.
*/
@Internal
-public class PulsarOrderedFetcherManager<T> extends
PulsarFetcherManagerBase<T> {
+public class PulsarOrderedFetcherManager extends PulsarFetcherManagerBase {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarOrderedFetcherManager.class);
public PulsarOrderedFetcherManager(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<T>>>
elementsQueue,
- Supplier<SplitReader<PulsarMessage<T>, PulsarPartitionSplit>>
splitReaderSupplier,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>>
splitReaderSupplier,
Configuration configuration) {
super(elementsQueue, splitReaderSupplier, configuration);
}
@@ -59,18 +57,18 @@ public class PulsarOrderedFetcherManager<T> extends
PulsarFetcherManagerBase<T>
LOG.debug("Acknowledge messages {}", cursorsToCommit);
cursorsToCommit.forEach(
(partition, messageId) -> {
- SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit>
fetcher =
+ SplitFetcher<Message<byte[]>, PulsarPartitionSplit>
fetcher =
getOrCreateFetcher(partition.toString());
triggerAcknowledge(fetcher, partition, messageId);
});
}
private void triggerAcknowledge(
- SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> splitFetcher,
+ SplitFetcher<Message<byte[]>, PulsarPartitionSplit> splitFetcher,
TopicPartition partition,
MessageId messageId) {
- PulsarOrderedPartitionSplitReader<T> splitReader =
- (PulsarOrderedPartitionSplitReader<T>)
splitFetcher.getSplitReader();
+ PulsarOrderedPartitionSplitReader splitReader =
+ (PulsarOrderedPartitionSplitReader)
splitFetcher.getSplitReader();
splitReader.notifyCheckpointComplete(partition, messageId);
startFetcher(splitFetcher);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java
index 449e992..33b811c 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java
@@ -24,12 +24,12 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarUnorderedPartitionSplitReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
import java.util.ArrayList;
import java.util.List;
@@ -42,15 +42,13 @@ import static java.util.stream.Collectors.toCollection;
* Pulsar's FetcherManager implementation for unordered consuming. This class
is needed to help
* acknowledge the message to Pulsar using the {@link Consumer} inside the
{@link
* PulsarUnorderedPartitionSplitReader}.
- *
- * @param <T> The message type for pulsar decoded message.
*/
@Internal
-public class PulsarUnorderedFetcherManager<T> extends
PulsarFetcherManagerBase<T> {
+public class PulsarUnorderedFetcherManager extends PulsarFetcherManagerBase {
public PulsarUnorderedFetcherManager(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<T>>>
elementsQueue,
- Supplier<SplitReader<PulsarMessage<T>, PulsarPartitionSplit>>
splitReaderSupplier,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>>
splitReaderSupplier,
Configuration configuration) {
super(elementsQueue, splitReaderSupplier, configuration);
}
@@ -65,8 +63,8 @@ public class PulsarUnorderedFetcherManager<T> extends
PulsarFetcherManagerBase<T
}
private Optional<PulsarPartitionSplit> snapshotReader(
- SplitReader<PulsarMessage<T>, PulsarPartitionSplit> splitReader) {
- return ((PulsarUnorderedPartitionSplitReader<T>) splitReader)
+ SplitReader<Message<byte[]>, PulsarPartitionSplit> splitReader) {
+ return ((PulsarUnorderedPartitionSplitReader) splitReader)
.snapshotState()
.map(PulsarPartitionSplitState::toPulsarPartitionSplit);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessage.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessage.java
deleted file mode 100644
index 0632e22..0000000
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessage.java
+++ /dev/null
@@ -1,74 +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.source.reader.message;
-
-import org.apache.flink.annotation.Internal;
-import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
-
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
-
-/**
- * The message instance that contains the required information which would be
used for committing
- * the consuming status.
- */
-@Internal
-public class PulsarMessage<T> {
-
- /**
- * The id of a given message. This id could be same for multiple {@link
PulsarMessage}, although
- * it is unique for every {@link Message}.
- */
- private final MessageId id;
-
- /** The value which deserialized by {@link PulsarDeserializationSchema}. */
- private final T value;
-
- /** The produce time for this message, it's a event time. */
- private final long eventTime;
-
- public PulsarMessage(MessageId id, T value, long eventTime) {
- this.id = id;
- this.value = value;
- this.eventTime = eventTime;
- }
-
- public MessageId getId() {
- return id;
- }
-
- public T getValue() {
- return value;
- }
-
- public long getEventTime() {
- return eventTime;
- }
-
- @Override
- public String toString() {
- return "PulsarMessage{"
- + "id="
- + id
- + ", value="
- + value
- + ", eventTime="
- + eventTime
- + '}';
- }
-}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessageCollector.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessageCollector.java
deleted file mode 100644
index f201425..0000000
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/message/PulsarMessageCollector.java
+++ /dev/null
@@ -1,60 +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.source.reader.message;
-
-import org.apache.flink.connector.base.source.reader.RecordsBySplits;
-import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
-import org.apache.flink.util.Collector;
-
-import org.apache.pulsar.client.api.Message;
-
-/**
- * This collector supplier is providing the {@link Collector} for accepting
the deserialized {@link
- * PulsarMessage} from pulsar {@link PulsarDeserializationSchema}.
- *
- * @param <T> The deserialized pulsar message type, aka the source message
type.
- */
-public class PulsarMessageCollector<T> implements Collector<T> {
-
- private final String splitId;
- private final RecordsBySplits.Builder<PulsarMessage<T>> builder;
- private Message<?> message;
-
- public PulsarMessageCollector(
- String splitId, RecordsBySplits.Builder<PulsarMessage<T>> builder)
{
- this.splitId = splitId;
- this.builder = builder;
- }
-
- public void setMessage(Message<?> message) {
- this.message = message;
- }
-
- @Override
- public void collect(T t) {
- PulsarMessage<T> result =
- new PulsarMessage<>(message.getMessageId(), t,
message.getEventTime());
- builder.add(splitId, result);
- }
-
- @Override
- public void close() {
- // Nothing to do for this collector.
- }
-}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
index 6ff0466..046575e 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
@@ -26,14 +26,15 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import
org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter;
import
org.apache.flink.connector.pulsar.source.reader.fetcher.PulsarOrderedFetcherManager;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarOrderedPartitionSplitReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
import org.apache.flink.core.io.InputStatus;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClient;
import org.slf4j.Logger;
@@ -67,16 +68,18 @@ public class PulsarOrderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT>
private ScheduledExecutorService cursorScheduler;
public PulsarOrderedSourceReader(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<OUT>>>
elementsQueue,
- Supplier<PulsarOrderedPartitionSplitReader<OUT>>
splitReaderSupplier,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ Supplier<PulsarOrderedPartitionSplitReader> splitReaderSupplier,
+ PulsarRecordEmitter<OUT> recordEmitter,
SourceReaderContext context,
SourceConfiguration sourceConfiguration,
PulsarClient pulsarClient,
PulsarAdmin pulsarAdmin) {
super(
elementsQueue,
- new PulsarOrderedFetcherManager<>(
+ new PulsarOrderedFetcherManager(
elementsQueue, splitReaderSupplier::get,
context.getConfiguration()),
+ recordEmitter,
context,
sourceConfiguration,
pulsarClient,
@@ -151,7 +154,7 @@ public class PulsarOrderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT>
LOG.debug("Committing cursors for checkpoint {}", checkpointId);
Map<TopicPartition, MessageId> cursors =
cursorsToCommit.get(checkpointId);
try {
- ((PulsarOrderedFetcherManager<OUT>)
splitFetcherManager).acknowledgeMessages(cursors);
+ ((PulsarOrderedFetcherManager)
splitFetcherManager).acknowledgeMessages(cursors);
LOG.debug("Successfully acknowledge cursors for checkpoint {}",
checkpointId);
// Clean up the cursors.
@@ -196,7 +199,7 @@ public class PulsarOrderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT>
}
try {
- ((PulsarOrderedFetcherManager<OUT>)
splitFetcherManager).acknowledgeMessages(cursors);
+ ((PulsarOrderedFetcherManager)
splitFetcherManager).acknowledgeMessages(cursors);
// Clean up the finish splits.
cursorsOfFinishedSplits.keySet().removeAll(cursors.keySet());
} catch (Exception e) {
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
index eee6950..d7fe471 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
@@ -25,11 +25,11 @@ import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompl
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import
org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter;
import
org.apache.flink.connector.pulsar.source.reader.fetcher.PulsarFetcherManagerBase;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import java.util.Collection;
@@ -42,25 +42,21 @@ import java.util.Set;
*/
abstract class PulsarSourceReaderBase<OUT>
extends SourceReaderBase<
- PulsarMessage<OUT>, OUT, PulsarPartitionSplit,
PulsarPartitionSplitState> {
+ Message<byte[]>, OUT, PulsarPartitionSplit,
PulsarPartitionSplitState> {
protected final SourceConfiguration sourceConfiguration;
protected final PulsarClient pulsarClient;
protected final PulsarAdmin pulsarAdmin;
protected PulsarSourceReaderBase(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<OUT>>>
elementsQueue,
- PulsarFetcherManagerBase<OUT> splitFetcherManager,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ PulsarFetcherManagerBase splitFetcherManager,
+ PulsarRecordEmitter<OUT> recordEmitter,
SourceReaderContext context,
SourceConfiguration sourceConfiguration,
PulsarClient pulsarClient,
PulsarAdmin pulsarAdmin) {
- super(
- elementsQueue,
- splitFetcherManager,
- new PulsarRecordEmitter<>(),
- sourceConfiguration,
- context);
+ super(elementsQueue, splitFetcherManager, recordEmitter,
sourceConfiguration, context);
this.sourceConfiguration = sourceConfiguration;
this.pulsarClient = pulsarClient;
@@ -96,7 +92,7 @@ abstract class PulsarSourceReaderBase<OUT>
protected void closeFinishedSplits(Set<String> finishedSplitIds) {
for (String splitId : finishedSplitIds) {
- ((PulsarFetcherManagerBase<OUT>)
splitFetcherManager).closeFetcher(splitId);
+ ((PulsarFetcherManagerBase)
splitFetcherManager).closeFetcher(splitId);
}
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
index 3bb4bc6..41a9b28 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
@@ -23,13 +23,14 @@ import
org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import
org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter;
import
org.apache.flink.connector.pulsar.source.reader.fetcher.PulsarUnorderedFetcherManager;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarUnorderedPartitionSplitReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import org.apache.pulsar.client.api.transaction.TxnID;
@@ -63,8 +64,9 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
private boolean started = false;
public PulsarUnorderedSourceReader(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<PulsarMessage<OUT>>>
elementsQueue,
- Supplier<PulsarUnorderedPartitionSplitReader<OUT>>
splitReaderSupplier,
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
+ Supplier<PulsarUnorderedPartitionSplitReader> splitReaderSupplier,
+ PulsarRecordEmitter<OUT> recordEmitter,
SourceReaderContext context,
SourceConfiguration sourceConfiguration,
PulsarClient pulsarClient,
@@ -72,8 +74,9 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
@Nullable TransactionCoordinatorClient coordinatorClient) {
super(
elementsQueue,
- new PulsarUnorderedFetcherManager<>(
+ new PulsarUnorderedFetcherManager(
elementsQueue, splitReaderSupplier::get,
context.getConfiguration()),
+ recordEmitter,
context,
sourceConfiguration,
pulsarClient,
@@ -142,7 +145,7 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
public List<PulsarPartitionSplit> snapshotState(long checkpointId) {
LOG.debug("Trigger the new transaction for downstream readers.");
List<PulsarPartitionSplit> splits =
- ((PulsarUnorderedFetcherManager<OUT>)
splitFetcherManager).snapshotState();
+ ((PulsarUnorderedFetcherManager)
splitFetcherManager).snapshotState();
if (coordinatorClient != null) {
// Snapshot the transaction status and commit it after checkpoint
finishing.
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
index 316431a..6b02e0f 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
@@ -21,7 +21,6 @@ package org.apache.flink.connector.pulsar.source.reader.split;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarOrderedSourceReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
@@ -45,20 +44,17 @@ import static
org.apache.flink.connector.pulsar.source.enumerator.cursor.Message
/**
* The split reader a given {@link PulsarPartitionSplit}, it would be closed
once the {@link
* PulsarOrderedSourceReader} is closed.
- *
- * @param <OUT> the type of the pulsar source message that would be serialized
to downstream.
*/
@Internal
-public class PulsarOrderedPartitionSplitReader<OUT> extends
PulsarPartitionSplitReaderBase<OUT> {
+public class PulsarOrderedPartitionSplitReader extends
PulsarPartitionSplitReaderBase {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarOrderedPartitionSplitReader.class);
public PulsarOrderedPartitionSplitReader(
PulsarClient pulsarClient,
PulsarAdmin pulsarAdmin,
- SourceConfiguration sourceConfiguration,
- PulsarDeserializationSchema<OUT> deserializationSchema) {
- super(pulsarClient, pulsarAdmin, sourceConfiguration,
deserializationSchema);
+ SourceConfiguration sourceConfiguration) {
+ super(pulsarClient, pulsarAdmin, sourceConfiguration);
}
@Override
@@ -70,9 +66,6 @@ public class PulsarOrderedPartitionSplitReader<OUT> extends
PulsarPartitionSplit
protected void finishedPollMessage(Message<byte[]> message) {
// Nothing to do here.
LOG.debug("Finished polling message {}", message);
-
- // Release message
- message.release();
}
@Override
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
index b884c57..08e9977 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
@@ -28,9 +28,6 @@ import
org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import
org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor.StopCondition;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
-import
org.apache.flink.connector.pulsar.source.reader.message.PulsarMessageCollector;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.util.Preconditions;
@@ -60,19 +57,14 @@ import static
org.apache.flink.connector.pulsar.source.config.PulsarSourceConfig
import static
org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator.KeySharedMode.JOIN;
import static org.apache.pulsar.client.api.KeySharedPolicy.stickyHashRange;
-/**
- * The common partition split reader.
- *
- * @param <OUT> the type of the pulsar source message that would be serialized
to downstream.
- */
-abstract class PulsarPartitionSplitReaderBase<OUT>
- implements SplitReader<PulsarMessage<OUT>, PulsarPartitionSplit> {
+/** The common partition split reader. */
+abstract class PulsarPartitionSplitReaderBase
+ implements SplitReader<Message<byte[]>, PulsarPartitionSplit> {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarPartitionSplitReaderBase.class);
protected final PulsarClient pulsarClient;
protected final PulsarAdmin pulsarAdmin;
protected final SourceConfiguration sourceConfiguration;
- protected final PulsarDeserializationSchema<OUT> deserializationSchema;
protected Consumer<byte[]> pulsarConsumer;
protected PulsarPartitionSplit registeredSplit;
@@ -80,17 +72,15 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
protected PulsarPartitionSplitReaderBase(
PulsarClient pulsarClient,
PulsarAdmin pulsarAdmin,
- SourceConfiguration sourceConfiguration,
- PulsarDeserializationSchema<OUT> deserializationSchema) {
+ SourceConfiguration sourceConfiguration) {
this.pulsarClient = pulsarClient;
this.pulsarAdmin = pulsarAdmin;
this.sourceConfiguration = sourceConfiguration;
- this.deserializationSchema = deserializationSchema;
}
@Override
- public RecordsWithSplitIds<PulsarMessage<OUT>> fetch() throws IOException {
- RecordsBySplits.Builder<PulsarMessage<OUT>> builder = new
RecordsBySplits.Builder<>();
+ public RecordsWithSplitIds<Message<byte[]>> fetch() throws IOException {
+ RecordsBySplits.Builder<Message<byte[]>> builder = new
RecordsBySplits.Builder<>();
// Return when no split registered to this reader.
if (pulsarConsumer == null || registeredSplit == null) {
@@ -99,10 +89,9 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
StopCursor stopCursor = registeredSplit.getStopCursor();
String splitId = registeredSplit.splitId();
- PulsarMessageCollector<OUT> collector = new
PulsarMessageCollector<>(splitId, builder);
Deadline deadline =
Deadline.fromNow(sourceConfiguration.getMaxFetchTime());
- // Consume message from pulsar until it was woke up by flink reader.
+ // Consume messages from pulsar until it was woken up by flink reader.
for (int messageNum = 0;
messageNum < sourceConfiguration.getMaxFetchRecords() &&
deadline.hasTimeLeft();
messageNum++) {
@@ -116,11 +105,9 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
StopCondition condition = stopCursor.shouldStop(message);
if (condition == StopCondition.CONTINUE || condition ==
StopCondition.EXACTLY) {
- // Deserialize message.
- collector.setMessage(message);
- deserializationSchema.deserialize(message, collector);
-
- // Acknowledge message if need.
+ // Collect original message.
+ builder.add(splitId, message);
+ // Acknowledge the message if you need.
finishedPollMessage(message);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
index 02fd8d9..f095c12 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.source.reader.split;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
-import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import
org.apache.flink.connector.pulsar.source.reader.source.PulsarUnorderedSourceReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import
org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
@@ -52,11 +51,9 @@ import static
org.apache.flink.connector.pulsar.common.utils.PulsarTransactionUt
/**
* The split reader a given {@link PulsarPartitionSplit}, it would be closed
once the {@link
* PulsarUnorderedSourceReader} is closed.
- *
- * @param <OUT> the type of the pulsar source message that would be serialized
to downstream.
*/
@Internal
-public class PulsarUnorderedPartitionSplitReader<OUT> extends
PulsarPartitionSplitReaderBase<OUT> {
+public class PulsarUnorderedPartitionSplitReader extends
PulsarPartitionSplitReaderBase {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarUnorderedPartitionSplitReader.class);
@@ -68,9 +65,8 @@ public class PulsarUnorderedPartitionSplitReader<OUT> extends
PulsarPartitionSpl
PulsarClient pulsarClient,
PulsarAdmin pulsarAdmin,
SourceConfiguration sourceConfiguration,
- PulsarDeserializationSchema<OUT> deserializationSchema,
TransactionCoordinatorClient coordinatorClient) {
- super(pulsarClient, pulsarAdmin, sourceConfiguration,
deserializationSchema);
+ super(pulsarClient, pulsarAdmin, sourceConfiguration);
this.coordinatorClient = coordinatorClient;
}
@@ -111,9 +107,6 @@ public class PulsarUnorderedPartitionSplitReader<OUT>
extends PulsarPartitionSpl
if (sourceConfiguration.isEnableAutoAcknowledgeMessage()) {
sneakyClient(() -> pulsarConsumer.acknowledge(message));
}
-
- // Release message
- message.release();
}
@Override
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
index 831d9e0..0aee463 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
@@ -18,17 +18,16 @@
package org.apache.flink.connector.pulsar.source.enumerator.cursor;
-import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
import
org.apache.flink.connector.pulsar.source.reader.split.PulsarOrderedPartitionSplitReader;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Test;
@@ -43,7 +42,6 @@ import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_TIME;
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
import static
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
-import static
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.flinkSchema;
import static org.assertj.core.api.Assertions.assertThat;
/** Test different implementation of StopCursor. */
@@ -54,12 +52,9 @@ class StopCursorTest extends PulsarTestSuiteBase {
String topicName = randomAlphanumeric(5);
operator().createTopic(topicName, 2);
- PulsarOrderedPartitionSplitReader<String> splitReader =
- new PulsarOrderedPartitionSplitReader<>(
- operator().client(),
- operator().admin(),
- sourceConfig(),
- flinkSchema(new SimpleStringSchema()));
+ PulsarOrderedPartitionSplitReader splitReader =
+ new PulsarOrderedPartitionSplitReader(
+ operator().client(), operator().admin(),
sourceConfig());
// send the first message and set the stopCursor to filter any late
stopCursor
operator()
.sendMessage(
@@ -77,7 +72,7 @@ class StopCursorTest extends PulsarTestSuiteBase {
SplitsAddition<PulsarPartitionSplit> addition = new
SplitsAddition<>(singletonList(split));
splitReader.handleSplitsChanges(addition);
// first fetch should have result
- RecordsWithSplitIds<PulsarMessage<String>> firstResult =
splitReader.fetch();
+ RecordsWithSplitIds<Message<byte[]>> firstResult = splitReader.fetch();
assertThat(firstResult.nextSplit()).isNotNull();
assertThat(firstResult.nextRecordFromSplit()).isNotNull();
assertThat(firstResult.finishedSplits()).isEmpty();
@@ -87,7 +82,7 @@ class StopCursorTest extends PulsarTestSuiteBase {
topicNameWithPartition(topicName, 0),
Schema.STRING,
randomAlphanumeric(10));
- RecordsWithSplitIds<PulsarMessage<String>> secondResult =
splitReader.fetch();
+ RecordsWithSplitIds<Message<byte[]>> secondResult =
splitReader.fetch();
assertThat(secondResult.nextSplit()).isNull();
assertThat(secondResult.finishedSplits()).isNotEmpty();
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReaderTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReaderTest.java
index 3d58d5e..4d93ce9 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReaderTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReaderTest.java
@@ -38,7 +38,7 @@ class PulsarOrderedPartitionSplitReaderTest extends
PulsarPartitionSplitReaderTe
@TestTemplate
void consumeMessageCreatedBeforeHandleSplitsChangesWithoutSeek(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
handleSplit(splitReader, topicName, 0);
@@ -47,7 +47,7 @@ class PulsarOrderedPartitionSplitReaderTest extends
PulsarPartitionSplitReaderTe
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndUseLatestStartCursorWithoutSeek(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
handleSplit(splitReader, topicName, 0, MessageId.latest);
@@ -56,7 +56,7 @@ class PulsarOrderedPartitionSplitReaderTest extends
PulsarPartitionSplitReaderTe
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndUseEarliestStartCursorWithoutSeek(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
handleSplit(splitReader, topicName, 0, MessageId.earliest);
@@ -65,7 +65,7 @@ class PulsarOrderedPartitionSplitReaderTest extends
PulsarPartitionSplitReaderTe
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndUseSecondLastMessageWithoutSeek(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
MessageIdImpl lastMessageId =
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
index e63424a..4631798 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
@@ -18,7 +18,6 @@
package org.apache.flink.connector.pulsar.source.reader.split;
-import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -26,13 +25,13 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
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.reader.message.PulsarMessage;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import
org.apache.flink.connector.pulsar.testutils.extension.TestOrderlinessExtension;
import org.apache.flink.util.TestLoggerExtension;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -65,7 +64,6 @@ import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_TIME;
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
import static
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
-import static
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.flinkSchema;
import static
org.apache.flink.connector.pulsar.testutils.PulsarTestCommonUtils.isAssignableFromParameterContext;
import static
org.apache.flink.connector.pulsar.testutils.extension.TestOrderlinessExtension.PULSAR_SOURCE_READER_SUBSCRIPTION_TYPE_STORE_KEY;
import static
org.apache.flink.connector.pulsar.testutils.extension.TestOrderlinessExtension.PULSAR_TEST_RESOURCE_NAMESPACE;
@@ -100,12 +98,12 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
protected void handleSplit(
- PulsarPartitionSplitReaderBase<String> reader, String topicName,
int partitionId) {
+ PulsarPartitionSplitReaderBase reader, String topicName, int
partitionId) {
handleSplit(reader, topicName, partitionId, null);
}
protected void handleSplit(
- PulsarPartitionSplitReaderBase<String> reader,
+ PulsarPartitionSplitReaderBase reader,
String topicName,
int partitionId,
MessageId startPosition) {
@@ -117,12 +115,12 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
private void seekStartPositionAndHandleSplit(
- PulsarPartitionSplitReaderBase<String> reader, String topicName,
int partitionId) {
+ PulsarPartitionSplitReaderBase reader, String topicName, int
partitionId) {
seekStartPositionAndHandleSplit(reader, topicName, partitionId,
MessageId.latest);
}
private void seekStartPositionAndHandleSplit(
- PulsarPartitionSplitReaderBase<String> reader,
+ PulsarPartitionSplitReaderBase reader,
String topicName,
int partitionId,
MessageId startPosition) {
@@ -159,29 +157,29 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
reader.handleSplitsChanges(addition);
}
- private <T> PulsarMessage<T>
fetchedMessage(PulsarPartitionSplitReaderBase<T> splitReader) {
+ private <T> Message<byte[]> fetchedMessage(PulsarPartitionSplitReaderBase
splitReader) {
return fetchedMessages(splitReader, 1,
false).stream().findFirst().orElse(null);
}
- protected <T> List<PulsarMessage<T>> fetchedMessages(
- PulsarPartitionSplitReaderBase<T> splitReader, int expectedCount,
boolean verify) {
+ protected <T> List<Message<byte[]>> fetchedMessages(
+ PulsarPartitionSplitReaderBase splitReader, int expectedCount,
boolean verify) {
return fetchedMessages(
splitReader, expectedCount, verify,
Boundedness.CONTINUOUS_UNBOUNDED);
}
- private <T> List<PulsarMessage<T>> fetchedMessages(
- PulsarPartitionSplitReaderBase<T> splitReader,
+ private <T> List<Message<byte[]>> fetchedMessages(
+ PulsarPartitionSplitReaderBase splitReader,
int expectedCount,
boolean verify,
Boundedness boundedness) {
- List<PulsarMessage<T>> messages = new ArrayList<>(expectedCount);
+ List<Message<byte[]>> messages = new ArrayList<>(expectedCount);
List<String> finishedSplits = new ArrayList<>();
for (int i = 0; i < 3; ) {
try {
- RecordsWithSplitIds<PulsarMessage<T>> recordsBySplitIds =
splitReader.fetch();
+ RecordsWithSplitIds<Message<byte[]>> recordsBySplitIds =
splitReader.fetch();
if (recordsBySplitIds.nextSplit() != null) {
// Collect the records in this split.
- PulsarMessage<T> record;
+ Message<byte[]> record;
while ((record = recordsBySplitIds.nextRecordFromSplit())
!= null) {
messages.add(record);
}
@@ -207,7 +205,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
@TestTemplate
- void pollMessageAfterTimeout(PulsarPartitionSplitReaderBase<String>
splitReader)
+ void pollMessageAfterTimeout(PulsarPartitionSplitReaderBase splitReader)
throws InterruptedException, TimeoutException {
String topicName = randomAlphabetic(10);
@@ -215,7 +213,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
handleSplit(splitReader, topicName, 0, MessageId.latest);
// Poll once with a null message
- PulsarMessage<String> message1 = fetchedMessage(splitReader);
+ Message<byte[]> message1 = fetchedMessage(splitReader);
assertThat(message1).isNull();
// Send a message to pulsar
@@ -225,7 +223,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
// Poll this message again
waitUtil(
() -> {
- PulsarMessage<String> message2 =
fetchedMessage(splitReader);
+ Message<byte[]> message2 = fetchedMessage(splitReader);
return message2 != null;
},
ofSeconds(Integer.MAX_VALUE),
@@ -234,7 +232,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
@TestTemplate
void consumeMessageCreatedAfterHandleSplitChangesAndFetch(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
handleSplit(splitReader, topicName, 0, MessageId.latest);
operator().sendMessage(topicNameWithPartition(topicName, 0), STRING,
randomAlphabetic(10));
@@ -243,7 +241,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
@TestTemplate
void consumeMessageCreatedBeforeHandleSplitsChanges(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
seekStartPositionAndHandleSplit(splitReader, topicName, 0);
@@ -252,7 +250,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndResetToEarliestPosition(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
seekStartPositionAndHandleSplit(splitReader, topicName, 0,
MessageId.earliest);
@@ -261,7 +259,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndResetToLatestPosition(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
seekStartPositionAndHandleSplit(splitReader, topicName, 0,
MessageId.latest);
@@ -270,7 +268,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
@TestTemplate
void
consumeMessageCreatedBeforeHandleSplitsChangesAndUseSecondLastMessageIdCursor(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().setupTopic(topicName, STRING, () -> randomAlphabetic(10));
@@ -296,7 +294,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
@TestTemplate
- void emptyTopic(PulsarPartitionSplitReaderBase<String> splitReader) {
+ void emptyTopic(PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().createTopic(topicName, DEFAULT_PARTITIONS);
seekStartPositionAndHandleSplit(splitReader, topicName, 0);
@@ -304,7 +302,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
@TestTemplate
- void emptyTopicWithoutSeek(PulsarPartitionSplitReaderBase<String>
splitReader) {
+ void emptyTopicWithoutSeek(PulsarPartitionSplitReaderBase splitReader) {
String topicName = randomAlphabetic(10);
operator().createTopic(topicName, DEFAULT_PARTITIONS);
handleSplit(splitReader, topicName, 0);
@@ -312,8 +310,7 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
@TestTemplate
- void wakeupSplitReaderShouldNotCauseException(
- PulsarPartitionSplitReaderBase<String> splitReader) {
+ void
wakeupSplitReaderShouldNotCauseException(PulsarPartitionSplitReaderBase
splitReader) {
handleSplit(splitReader, "non-exist", 0);
AtomicReference<Throwable> error = new AtomicReference<>();
Thread t =
@@ -336,25 +333,18 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
}
@TestTemplate
- void assignNoSplits(PulsarPartitionSplitReaderBase<String> splitReader) {
+ void assignNoSplits(PulsarPartitionSplitReaderBase splitReader) {
assertThat(fetchedMessage(splitReader)).isNull();
}
/** Create a split reader with max message 1, fetch timeout 1s. */
- private PulsarPartitionSplitReaderBase<String>
splitReader(SubscriptionType subscriptionType) {
+ private PulsarPartitionSplitReaderBase splitReader(SubscriptionType
subscriptionType) {
if (subscriptionType == SubscriptionType.Failover) {
- return new PulsarOrderedPartitionSplitReader<>(
- operator().client(),
- operator().admin(),
- sourceConfig(),
- flinkSchema(new SimpleStringSchema()));
+ return new PulsarOrderedPartitionSplitReader(
+ operator().client(), operator().admin(), sourceConfig());
} else {
- return new PulsarUnorderedPartitionSplitReader<>(
- operator().client(),
- operator().admin(),
- sourceConfig(),
- flinkSchema(new SimpleStringSchema()),
- null);
+ return new PulsarUnorderedPartitionSplitReader(
+ operator().client(), operator().admin(), sourceConfig(),
null);
}
}
@@ -382,9 +372,9 @@ abstract class PulsarPartitionSplitReaderTestBase extends
PulsarTestSuiteBase {
public static class PulsarSplitReaderInvocationContext
implements TestTemplateInvocationContext {
- private final PulsarPartitionSplitReaderBase<?> splitReader;
+ private final PulsarPartitionSplitReaderBase splitReader;
- public
PulsarSplitReaderInvocationContext(PulsarPartitionSplitReaderBase<?>
splitReader) {
+ public
PulsarSplitReaderInvocationContext(PulsarPartitionSplitReaderBase splitReader) {
this.splitReader = checkNotNull(splitReader);
}
