This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
The following commit(s) were added to refs/heads/main by this push:
new 3e0a964 [FLINK-40676] Support Flink 2.0 and Pulsar Client 4.0.9
3e0a964 is described below
commit 3e0a9641ce4fd81076b274fb6e659a5734332f45
Author: fengyubiao <[email protected]>
AuthorDate: Wed Sep 16 01:32:30 2026 +0800
[FLINK-40676] Support Flink 2.0 and Pulsar Client 4.0.9
---
.github/workflows/push_pr.yml | 4 +-
.github/workflows/weekly.yml | 10 +-
flink-connector-pulsar-e2e-tests/pom.xml | 6 +-
.../tests/util/pulsar/PulsarSourceE2ECase.java | 36 +++
flink-connector-pulsar/pom.xml | 31 ++-
.../pulsar/common/config/PulsarConfiguration.java | 4 +-
.../common/schema/PulsarSchemaTypeInformation.java | 4 +-
.../common/schema/PulsarSchemaTypeSerializer.java | 2 +-
.../pulsar/common/utils/PulsarSerdeUtils.java | 47 ++++
.../flink/connector/pulsar/sink/PulsarSink.java | 17 +-
.../pulsar/sink/committer/PulsarCommitter.java | 4 +-
.../pulsar/sink/config/SinkConfiguration.java | 1 -
.../connector/pulsar/sink/writer/PulsarWriter.java | 8 +-
.../sink/writer/context/PulsarSinkContextImpl.java | 8 +-
.../pulsar/sink/writer/topic/MetadataListener.java | 5 +-
.../source/enumerator/cursor/CursorPosition.java | 25 ++
.../subscriber/impl/BasePulsarSubscriber.java | 3 +-
.../source/reader/PulsarPartitionSplitReader.java | 79 +++++-
.../source/reader/PulsarSourceFetcherManager.java | 9 +-
.../pulsar/source/reader/PulsarSourceReader.java | 28 ++-
.../PulsarDeserializationSchemaWrapper.java | 22 +-
.../reader/deserializer/PulsarSchemaWrapper.java | 19 +-
.../deserializer/PulsarTypeInformationWrapper.java | 28 ++-
.../PulsarTableSerializationSchemaFactory.java | 5 +-
.../PulsarTableDeserializationSchemaFactory.java | 5 +-
.../pulsar/common/MiniClusterTestEnvironment.java | 6 +-
.../schema/PulsarSchemaTypeInformationTest.java | 2 +-
.../schema/factories/AvroSchemaFactoryTest.java | 2 +-
.../pulsar/sink/writer/PulsarWriterTest.java | 40 +--
.../sink/writer/topic/ProducerRegisterTest.java | 10 +-
.../pulsar/source/PulsarSourceITCase.java | 37 +++
.../reader/PulsarPartitionSplitReaderTest.java | 22 +-
.../source/reader/PulsarSourceReaderTest.java | 84 ++++---
.../connector/pulsar/table/PulsarTableITCase.java | 2 +-
.../pulsar/table/PulsarTableTestBase.java | 24 +-
.../testutils/SimpleCollectIteratorAssert.java | 279 +++++++++++++++++++++
.../pulsar/testutils/function/ControlSource.java | 2 +-
.../runtime/container/PulsarContainerRuntime.java | 2 +-
flink-sql-connector-pulsar/pom.xml | 11 +-
.../src/main/resources/META-INF/NOTICE | 19 +-
pom.xml | 46 +++-
41 files changed, 828 insertions(+), 170 deletions(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index 8905155..55d5bf6 100644
--- a/.github/workflows/push_pr.yml
+++ b/.github/workflows/push_pr.yml
@@ -28,8 +28,8 @@ jobs:
compile_and_test:
strategy:
matrix:
- flink: [ 1.20.4 ]
- jdk: [ '8, 11, 17' ]
+ flink: [ 2.0.2 ]
+ jdk: [ '11, 17' ]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink }}
diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
index 4f54b40..75cc359 100644
--- a/.github/workflows/weekly.yml
+++ b/.github/workflows/weekly.yml
@@ -31,18 +31,20 @@ jobs:
strategy:
matrix:
flink_branches: [{
- flink: 1.20-SNAPSHOT,
- branch: main
+ flink: 2.0.2,
+ branch: main,
+ jdk: "11,17"
}, {
flink: 1.20.4,
- branch: v4.2
+ branch: v4.2,
+ jdk: "8,11,17"
}
]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink_branches.flink }}
connector_branch: ${{ matrix.flink_branches.branch }}
- jdk_version: ${{ matrix.flink_branches.jdk || '8, 11, 17' }}
+ jdk_version: ${{ matrix.flink_branches.jdk }}
run_dependency_convergence: false
timeout_global: 120
timeout_test: 80
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml
b/flink-connector-pulsar-e2e-tests/pom.xml
index 6e7c305..52acd3e 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -60,9 +60,13 @@ under the License.
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
- <artifactId>pulsar</artifactId>
+ <artifactId>testcontainers-pulsar</artifactId>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
diff --git
a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
index e3a54f0..f9a9835 100644
---
a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
+++
b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
@@ -18,6 +18,7 @@
package org.apache.flink.tests.util.pulsar;
+import org.apache.flink.connector.pulsar.testutils.SimpleCollectIteratorAssert;
import
org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
import
org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
import
org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
@@ -31,6 +32,11 @@ import
org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerUtils;
import
org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import static org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE;
@@ -72,4 +78,34 @@ public class PulsarSourceE2ECase extends
SourceTestSuiteBase<String> {
context.addConnectorJarPaths(FlinkContainerUtils.connectorJarPaths());
return context;
};
+
+ /**
+ * {@link CollectIteratorAssertions} will generate a mismatch description
if the result does not
+ * match the expected value. It attempts to capture all following
messages, even though already
+ * failed, which helps engineers for troubleshooting, but draining the
following messages may
+ * lead the test to get stuck. We rewrite the method to avoid the test to
get stuck.
+ */
+ @Override
+ protected void checkResultWithSemantic(
+ CloseableIterator<String> resultIterator,
+ List<List<String>> testData,
+ org.apache.flink.core.execution.CheckpointingMode semantic,
+ Integer limit) {
+ if (limit != null) {
+ Runnable runnable =
+ () ->
+ new SimpleCollectIteratorAssert<>(resultIterator)
+ .withNumRecordsLimit(limit)
+ .matchesRecordsFromSource(testData,
semantic);
+
+ try {
+ CompletableFuture.runAsync(runnable).get(65, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new AssertionError("Timed out or failed while validating
source results.", e);
+ }
+ } else {
+ new SimpleCollectIteratorAssert<>(resultIterator)
+ .matchesRecordsFromSource(testData, semantic);
+ }
+ }
}
diff --git a/flink-connector-pulsar/pom.xml b/flink-connector-pulsar/pom.xml
index 850366d..be0aed2 100644
--- a/flink-connector-pulsar/pom.xml
+++ b/flink-connector-pulsar/pom.xml
@@ -95,6 +95,21 @@ under the License.
<artifactId>pulsar-client-all</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api-incubator</artifactId>
+ </dependency>
+
<!-- Tests -->
<dependency>
@@ -119,6 +134,12 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<!-- Table factory testing -->
<dependency>
<groupId>org.apache.flink</groupId>
@@ -153,7 +174,7 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
- <artifactId>pulsar</artifactId>
+ <artifactId>testcontainers-pulsar</artifactId>
<scope>test</scope>
</dependency>
@@ -261,6 +282,14 @@ under the License.
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <!-- Avoid CI hanging indefinitely when
tests are done but forked JVM doesn't exit cleanly. -->
+
<forkedProcessExitTimeoutInSeconds>120</forkedProcessExitTimeoutInSeconds>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
index 0681a3e..30dc831 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
@@ -77,7 +77,7 @@ public abstract class PulsarConfiguration extends
UnmodifiableConfiguration {
}
/** Get an option value from the given config, convert it into a new value
instance. */
- public <F, T> T get(ConfigOption<F> option, Function<F, T> convertor) {
+ public <F, T> T getAndConvert(ConfigOption<F> option, Function<F, T>
convertor) {
F value = get(option);
if (value != null) {
return convertor.apply(value);
@@ -97,7 +97,7 @@ public abstract class PulsarConfiguration extends
UnmodifiableConfiguration {
public <T, V> void useOption(
ConfigOption<T> option, Function<T, V> convertor, Consumer<V>
setter) {
if (contains(option) || option.hasDefaultValue()) {
- V value = get(option, convertor);
+ V value = getAndConvert(option, convertor);
setter.accept(value);
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformation.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformation.java
index ef616f0..3843548 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformation.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformation.java
@@ -19,7 +19,7 @@
package org.apache.flink.connector.pulsar.common.schema;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.SerializerConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -67,7 +67,7 @@ public class PulsarSchemaTypeInformation<T> extends
TypeInformation<T> {
}
@Override
- public TypeSerializer<T> createSerializer(ExecutionConfig config) {
+ public TypeSerializer<T> createSerializer(SerializerConfig
serializerConfig) {
return new PulsarSchemaTypeSerializer<>(schema);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java
index ef5c9f0..c94a851 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java
@@ -202,7 +202,7 @@ public class PulsarSchemaTypeSerializer<T> extends
TypeSerializer<T> {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer) {
+ TypeSerializerSnapshot<T> typeSerializerSnapshot) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarSerdeUtils.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarSerdeUtils.java
index ffe041b..eda5cab 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarSerdeUtils.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarSerdeUtils.java
@@ -19,6 +19,11 @@
package org.apache.flink.connector.pulsar.common.utils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.BiConsumerWithException;
@@ -33,6 +38,10 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/** Util for serialize and deserialize. */
@Internal
@@ -158,4 +167,42 @@ public final class PulsarSerdeUtils {
}
return result;
}
+
+ /**
+ * Local replacement for Flink 1.18's DataTypeUtils.stripRowPrefix which
was removed in Flink
+ * 2.x.
+ */
+ public static DataType stripRowPrefix(DataType dataType, String prefix) {
+ checkArgument(dataType.getLogicalType().is(LogicalTypeRoot.ROW), "Row
data type expected.");
+ final RowType rowType = (RowType) dataType.getLogicalType();
+ final List<String> newFieldNames =
+ rowType.getFieldNames().stream()
+ .map(
+ name ->
+ name.startsWith(prefix)
+ ?
name.substring(prefix.length())
+ : name)
+ .collect(Collectors.toList());
+ final LogicalType newRowType = renameRowFields(rowType, newFieldNames);
+ return new FieldsDataType(
+ newRowType, dataType.getConversionClass(),
dataType.getChildren());
+ }
+
+ public static RowType renameRowFields(RowType rowType, List<String>
newFieldNames) {
+ Preconditions.checkArgument(
+ rowType.getFieldCount() == newFieldNames.size(),
+ "Row length and new names must match.");
+ final List<RowType.RowField> newFields =
+ IntStream.range(0, rowType.getFieldCount())
+ .mapToObj(
+ pos -> {
+ final RowType.RowField oldField =
rowType.getFields().get(pos);
+ return new RowType.RowField(
+ newFieldNames.get(pos),
+ oldField.getType(),
+
oldField.getDescription().orElse(null));
+ })
+ .collect(Collectors.toList());
+ return new RowType(rowType.isNullable(), newFields);
+ }
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
index 559da3a..7a6e303 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
@@ -21,7 +21,11 @@ package org.apache.flink.connector.pulsar.sink;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.connector.sink2.Committer;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
+import org.apache.flink.api.connector.sink2.CommitterInitContext;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SupportsCommitter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
@@ -42,6 +46,8 @@ import org.apache.pulsar.client.api.PulsarClientException;
import javax.annotation.Nullable;
+import java.io.IOException;
+
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -81,7 +87,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
* @param <IN> The input type of the sink.
*/
@PublicEvolving
-public class PulsarSink<IN> implements TwoPhaseCommittingSink<IN,
PulsarCommittable> {
+public class PulsarSink<IN> implements Sink<IN>,
SupportsCommitter<PulsarCommittable> {
private static final long serialVersionUID = 4416714587951282119L;
private final SinkConfiguration sinkConfiguration;
@@ -129,7 +135,7 @@ public class PulsarSink<IN> implements
TwoPhaseCommittingSink<IN, PulsarCommitta
@Internal
@Override
- public PrecommittingSinkWriter<IN, PulsarCommittable>
createWriter(InitContext initContext)
+ public CommittingSinkWriter<IN, PulsarCommittable>
createWriter(WriterInitContext initContext)
throws PulsarClientException {
return new PulsarWriter<>(
sinkConfiguration,
@@ -143,8 +149,9 @@ public class PulsarSink<IN> implements
TwoPhaseCommittingSink<IN, PulsarCommitta
@Internal
@Override
- public Committer<PulsarCommittable> createCommitter() {
- return new PulsarCommitter(sinkConfiguration);
+ public Committer<PulsarCommittable> createCommitter(CommitterInitContext
committerInitContext)
+ throws IOException {
+ return new PulsarCommitter(sinkConfiguration, committerInitContext);
}
@Internal
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
index c51598d..4b7506c 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.pulsar.sink.committer;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.api.connector.sink2.CommitterInitContext;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.sink.PulsarSink;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
@@ -60,7 +61,8 @@ public class PulsarCommitter implements
Committer<PulsarCommittable>, Closeable
private PulsarClient pulsarClient;
private TransactionCoordinatorClient coordinatorClient;
- public PulsarCommitter(SinkConfiguration sinkConfiguration) {
+ public PulsarCommitter(
+ SinkConfiguration sinkConfiguration, CommitterInitContext
committerInitContext) {
this.sinkConfiguration = checkNotNull(sinkConfiguration);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java
index 3bc7a42..c546420 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java
@@ -19,7 +19,6 @@
package org.apache.flink.connector.pulsar.sink.config;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.config.PulsarConfiguration;
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
index 5c93ef3..021a8f5 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
@@ -22,8 +22,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import
org.apache.flink.api.common.serialization.SerializationSchema.InitializationContext;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
-import
org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink.PrecommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
@@ -67,7 +67,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
* @param <IN> The type of the input elements.
*/
@Internal
-public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN,
PulsarCommittable> {
+public class PulsarWriter<IN> implements CommittingSinkWriter<IN,
PulsarCommittable> {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarWriter.class);
private final PulsarSerializationSchema<IN> serializationSchema;
@@ -101,7 +101,7 @@ public class PulsarWriter<IN> implements
PrecommittingSinkWriter<IN, PulsarCommi
TopicRouter<IN> topicRouter,
MessageDelayer<IN> messageDelayer,
PulsarCrypto pulsarCrypto,
- InitContext initContext)
+ WriterInitContext initContext)
throws PulsarClientException {
checkNotNull(sinkConfiguration);
this.serializationSchema = checkNotNull(serializationSchema);
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
index 8c827db..b1c5c96 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
@@ -20,7 +20,7 @@ package org.apache.flink.connector.pulsar.sink.writer.context;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
@@ -38,11 +38,11 @@ public class PulsarSinkContextImpl implements
PulsarSinkContext {
private final MetadataListener metadataListener;
public PulsarSinkContextImpl(
- InitContext initContext,
+ WriterInitContext initContext,
SinkConfiguration sinkConfiguration,
MetadataListener metadataListener) {
- this.parallelInstanceId = initContext.getSubtaskId();
- this.numberOfParallelSubtasks =
initContext.getNumberOfParallelSubtasks();
+ this.parallelInstanceId =
initContext.getTaskInfo().getIndexOfThisSubtask();
+ this.numberOfParallelSubtasks =
initContext.getTaskInfo().getNumberOfParallelSubtasks();
this.processingTimeService = initContext.getProcessingTimeService();
this.enableSchemaEvolution =
sinkConfiguration.isEnableSchemaEvolution();
this.metadataListener = metadataListener;
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
index b324efe..8b389d6 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
@@ -112,7 +112,10 @@ public class MetadataListener implements Serializable,
Closeable {
public Optional<Integer> load(String topic)
throws ExecutionException,
InterruptedException {
PartitionedTopicMetadata metadata =
-
clientImpl.getPartitionedTopicMetadata(topic).get();
+ clientImpl
+
.getPartitionedTopicMetadata(
+ topic, true,
true)
+ .get();
return
Optional.of(metadata.partitions);
}
});
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/CursorPosition.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/CursorPosition.java
index 8d92ac5..cf4f84b 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/CursorPosition.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/CursorPosition.java
@@ -32,6 +32,8 @@ import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.client.impl.ChunkMessageIdImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.Serializable;
@@ -43,6 +45,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
@PublicEvolving
public final class CursorPosition implements Serializable {
private static final long serialVersionUID = -802405183307684549L;
+ private static final Logger LOG =
LoggerFactory.getLogger(CursorPosition.class);
private final Type type;
@@ -91,11 +94,33 @@ public final class CursorPosition implements Serializable {
.subscribe()) {
// Reset cursor to desired position.
if (type == Type.TIMESTAMP) {
+ LOG.info(
+ "[{}] [{}] reset cursor to timestamp {}",
+ topicName,
+ subscriptionName,
+ this.timestamp);
consumer.seek(getActualTimestamp(this.timestamp));
} else if (messageId instanceof ChunkMessageIdImpl) {
MessageIdAdv msgId = ((ChunkMessageIdImpl)
messageId).getFirstChunkMessageId();
+ LOG.info(
+ "[{}] [{}] reset cursor to chunk msg id {}:{}:{}/{}",
+ topicName,
+ subscriptionName,
+ msgId.getLedgerId(),
+ msgId.getEntryId(),
+ msgId.getBatchIndex(),
+ msgId.getBatchSize());
consumer.seek(getActualMessageId(msgId));
} else {
+ MessageIdAdv msgId = (MessageIdAdv) messageId;
+ LOG.info(
+ "[{}] [{}] reset cursor to msg id {}:{}:{}/{}",
+ topicName,
+ subscriptionName,
+ msgId.getLedgerId(),
+ msgId.getEntryId(),
+ msgId.getBatchIndex(),
+ msgId.getBatchSize());
consumer.seek(getActualMessageId((MessageIdAdv) messageId));
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
index 30a24df..5d6c443 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
@@ -53,7 +53,8 @@ public abstract class BasePulsarSubscriber implements
PulsarSubscriber {
}
PulsarClientImpl clientImpl = (PulsarClientImpl) client;
- PartitionedTopicMetadata metadata =
clientImpl.getPartitionedTopicMetadata(topic).get();
+ PartitionedTopicMetadata metadata =
+ clientImpl.getPartitionedTopicMetadata(topic, true,
true).get();
if (metadata.partitions == NON_PARTITIONED) {
NON_PARTITIONED_TOPICS.add(topic);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java
index 93d5d95..19ea784 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java
@@ -46,6 +46,7 @@ import org.apache.pulsar.client.api.KeySharedPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageCrypto;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -57,8 +58,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -91,6 +94,27 @@ import static
org.apache.pulsar.client.api.KeySharedPolicy.stickyHashRange;
public class PulsarPartitionSplitReader
implements SplitReader<Message<byte[]>, PulsarPartitionSplit> {
private static final Logger LOG =
LoggerFactory.getLogger(PulsarPartitionSplitReader.class);
+ private static final Comparator<MessageIdAdv> MESSAGE_ID_COMPARATOR =
+ (messageId1, messageId2) -> {
+ int ledgerComparison =
+ Long.compare(messageId1.getLedgerId(),
messageId2.getLedgerId());
+ if (ledgerComparison != 0) {
+ return ledgerComparison;
+ }
+
+ int entryComparison =
+ Long.compare(messageId1.getEntryId(),
messageId2.getEntryId());
+ if (entryComparison != 0) {
+ return entryComparison;
+ }
+
+ if (messageId1 instanceof BatchMessageIdImpl
+ && messageId2 instanceof BatchMessageIdImpl) {
+ return Integer.compare(messageId1.getBatchIndex(),
messageId2.getBatchIndex());
+ }
+
+ return 0;
+ };
private final PulsarClient pulsarClient;
private final SourceConfiguration sourceConfiguration;
@@ -129,25 +153,66 @@ public class PulsarPartitionSplitReader
Deadline deadline =
Deadline.fromNow(sourceConfiguration.getMaxFetchTime());
// Consume messages from pulsar until it was woken up by flink reader.
+ CompletableFuture<Message<byte[]>> msgFuture = null;
+ MessageIdAdv latestMessageIdInTheCurrentFetch = null;
for (int messageNum = 0;
- messageNum < sourceConfiguration.getMaxFetchRecords() &&
deadline.hasTimeLeft();
- messageNum++) {
+ messageNum < sourceConfiguration.getMaxFetchRecords() &&
deadline.hasTimeLeft(); ) {
try {
int fetchTime = sourceConfiguration.getFetchOneMessageTime();
if (fetchTime <= 0) {
fetchTime = (int) deadline.timeLeftIfAny().toMillis();
}
-
- Message<byte[]> message = pulsarConsumer.receive(fetchTime,
TimeUnit.MILLISECONDS);
+ // (Highlight) The synchronised API "receive(Duration)" has a
bug, which may throw
+ // an error: "Try to
+ // reserve/release memory failed, the param memorySize is a
negative value".
+ // Here we use an asynchronous API.
+ msgFuture = pulsarConsumer.receiveAsync();
+ Message<byte[]> message = null;
+ try {
+ message = msgFuture.get(fetchTime, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ if (msgFuture.completeExceptionally(e)) {
+ throw e;
+ } else if (!msgFuture.isCompletedExceptionally()) {
+ message = msgFuture.get();
+ } else {
+ // throws error.
+ msgFuture.get();
+ }
+ }
if (message == null) {
break;
}
+ MessageIdAdv msgId = (MessageIdAdv) message.getMessageId();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "[{}] [{}] received a message {}:{}:{}/{}.",
+ pulsarConsumer.getTopic(),
+ pulsarConsumer.getSubscription(),
+ msgId.getLedgerId(),
+ msgId.getEntryId(),
+ msgId.getBatchIndex(),
+ msgId.getBatchSize());
+ }
+ // (Highlight) Since the connector will not acknowledge
messages immediately, when
+ // the pulsar consumer
+ // reconnects, it may receive repeated messages. We use the
following two mechanism
+ // to solve the
+ // repeated receiving messages issue.
+ if (isAtOrAfter(latestMessageIdInTheCurrentFetch, msgId)
+ || isAtOrAfter(
+ (MessageIdAdv)
registeredSplit.getLatestConsumedId(), msgId)) {
+ continue;
+ }
+ latestMessageIdInTheCurrentFetch = msgId;
+
StopCondition condition = stopCursor.shouldStop(message);
if (condition == StopCondition.CONTINUE || condition ==
StopCondition.EXACTLY) {
// Collect original message.
builder.add(splitId, message);
+ messageNum++;
LOG.debug("Finished polling message {}", message);
}
@@ -165,6 +230,12 @@ public class PulsarPartitionSplitReader
return builder.build();
}
+ /** Returns whether the previous message ID is at or after the given
message ID. */
+ private static boolean isAtOrAfter(MessageIdAdv previousMessageId,
MessageIdAdv messageId) {
+ return previousMessageId != null
+ && MESSAGE_ID_COMPARATOR.compare(previousMessageId, messageId)
>= 0;
+ }
+
@Override
public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit>
splitsChanges) {
LOG.debug("Handle split changes {}", splitsChanges);
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java
index 2680ba7..1430d13 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java
@@ -20,12 +20,9 @@ package org.apache.flink.connector.pulsar.source.reader;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
-import org.apache.flink.connector.base.source.reader.SourceReaderBase;
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.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
@@ -59,16 +56,12 @@ public class PulsarSourceFetcherManager
/**
* Creates a new SplitFetcherManager with multiple I/O threads.
*
- * @param elementsQueue The queue that is used to hand over data from the
I/O thread (the
- * fetchers) to the reader, which emits the records and book-keeps the
state. This must be
- * the same queue instance that is also passed to the {@link
SourceReaderBase}.
* @param splitReaderSupplier The factory for the split reader that
connects to the source
*/
public PulsarSourceFetcherManager(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>>
splitReaderSupplier,
Configuration configuration) {
- super(elementsQueue, splitReaderSupplier, configuration);
+ super(splitReaderSupplier, configuration);
}
/**
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReader.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReader.java
index 014d303..7c7c336 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReader.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReader.java
@@ -22,10 +22,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.ReaderOutput;
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.SourceReaderBase;
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.common.crypto.PulsarCrypto;
import org.apache.flink.connector.pulsar.common.schema.BytesSchema;
import org.apache.flink.connector.pulsar.common.schema.PulsarSchema;
@@ -41,6 +39,7 @@ import org.apache.flink.util.FlinkRuntimeException;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.slf4j.Logger;
@@ -84,14 +83,12 @@ public class PulsarSourceReader<OUT>
private ScheduledExecutorService cursorScheduler;
private PulsarSourceReader(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
PulsarSourceFetcherManager fetcherManager,
PulsarDeserializationSchema<OUT> deserializationSchema,
SourceConfiguration sourceConfiguration,
PulsarClient pulsarClient,
SourceReaderContext context) {
super(
- elementsQueue,
fetcherManager,
new PulsarRecordEmitter<>(deserializationSchema),
sourceConfiguration,
@@ -179,7 +176,22 @@ public class PulsarSourceReader<OUT>
for (PulsarPartitionSplit split : splits) {
MessageId latestConsumedId = split.getLatestConsumedId();
if (latestConsumedId != null) {
+ MessageIdAdv msgId = (MessageIdAdv) latestConsumedId;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "[{}] [{}] snapshot state for partition
{}:{}:{}/{}",
+ split.getPartition().getFullTopicName(),
+ sourceConfiguration.getSubscriptionName(),
+ msgId.getLedgerId(),
+ msgId.getEntryId(),
+ msgId.getBatchIndex(),
+ msgId.getBatchSize());
+ }
cursors.put(split.getPartition(), latestConsumedId);
+ } else {
+ LOG.warn(
+ "{} snapshot state for partition null",
+ split.getPartition().getFullTopicName());
}
}
// Put cursors of all the finished splits.
@@ -252,11 +264,6 @@ public class PulsarSourceReader<OUT>
SourceReaderContext readerContext)
throws Exception {
- // Create a message queue with the predefined source option.
- int queueCapacity = sourceConfiguration.getMessageQueueCapacity();
- FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue =
- new FutureCompletingBlockingQueue<>(queueCapacity);
-
PulsarClient pulsarClient = createClient(sourceConfiguration);
// Initialize the deserialization schema before creating the pulsar
reader.
@@ -287,10 +294,9 @@ public class PulsarSourceReader<OUT>
PulsarSourceFetcherManager fetcherManager =
new PulsarSourceFetcherManager(
- elementsQueue, splitReaderSupplier,
readerContext.getConfiguration());
+ splitReaderSupplier, readerContext.getConfiguration());
return new PulsarSourceReader<>(
- elementsQueue,
fetcherManager,
deserializationSchema,
sourceConfiguration,
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaWrapper.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaWrapper.java
index ca9b11f..365e177 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaWrapper.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaWrapper.java
@@ -25,6 +25,10 @@ import
org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.util.Collector;
import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A {@link PulsarDeserializationSchema} implementation which based on the
given flink's {@link
@@ -36,6 +40,8 @@ import org.apache.pulsar.client.api.Message;
@Internal
public class PulsarDeserializationSchemaWrapper<T> implements
PulsarDeserializationSchema<T> {
private static final long serialVersionUID = -630646912412751300L;
+ private static final Logger LOG =
+ LoggerFactory.getLogger(PulsarDeserializationSchemaWrapper.class);
private final DeserializationSchema<T> deserializationSchema;
@@ -52,9 +58,23 @@ public class PulsarDeserializationSchemaWrapper<T>
implements PulsarDeserializat
@Override
public void deserialize(Message<byte[]> message, Collector<T> out) throws
Exception {
+ MessageId msgId = message.getMessageId();
+ MessageIdAdv messageIdAdv = (MessageIdAdv) msgId;
byte[] bytes = message.getData();
T instance = deserializationSchema.deserialize(bytes);
-
+ if (LOG.isDebugEnabled()) {
+ if (messageIdAdv != null) {
+ LOG.debug(
+ "Deserialize message {}:{}:{}/{} of {}",
+ messageIdAdv.getLedgerId(),
+ messageIdAdv.getEntryId(),
+ messageIdAdv.getBatchIndex(),
+ messageIdAdv.getBatchSize(),
+ String.valueOf(instance));
+ } else {
+ LOG.debug("Deserialize message {id-null} of {}",
String.valueOf(instance));
+ }
+ }
out.collect(instance);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarSchemaWrapper.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarSchemaWrapper.java
index 37553a5..26954e3 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarSchemaWrapper.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarSchemaWrapper.java
@@ -23,9 +23,12 @@ import
org.apache.flink.connector.pulsar.common.schema.PulsarSchema;
import org.apache.flink.util.Collector;
import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static
org.apache.flink.connector.pulsar.common.schema.PulsarSchemaUtils.createTypeInformation;
@@ -38,6 +41,7 @@ import static
org.apache.flink.connector.pulsar.common.schema.PulsarSchemaUtils.
@Internal
public class PulsarSchemaWrapper<T> implements PulsarDeserializationSchema<T> {
private static final long serialVersionUID = -4864701207257059158L;
+ private static final Logger LOG =
LoggerFactory.getLogger(PulsarSchemaWrapper.class);
/** The serializable pulsar schema, it wrap the schema with type class. */
private final PulsarSchema<T> pulsarSchema;
@@ -64,7 +68,20 @@ public class PulsarSchemaWrapper<T> implements
PulsarDeserializationSchema<T> {
Schema<T> schema = this.pulsarSchema.getPulsarSchema();
byte[] bytes = message.getData();
T instance = schema.decode(bytes);
-
+ MessageIdAdv messageIdAdv = (MessageIdAdv) message.getMessageId();
+ if (LOG.isDebugEnabled()) {
+ if (messageIdAdv != null) {
+ LOG.debug(
+ "Deserialize message {}:{}:{}/{} of {}",
+ messageIdAdv.getLedgerId(),
+ messageIdAdv.getEntryId(),
+ messageIdAdv.getBatchIndex(),
+ messageIdAdv.getBatchSize(),
+ String.valueOf(instance));
+ } else {
+ LOG.debug("Deserialize message {id-null} of {}",
String.valueOf(instance));
+ }
+ }
out.collect(instance);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarTypeInformationWrapper.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarTypeInformationWrapper.java
index 59f77d7..4bfdd19 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarTypeInformationWrapper.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarTypeInformationWrapper.java
@@ -21,11 +21,15 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.util.Collector;
import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
/**
* Wrap the flink TypeInformation into a {@code PulsarDeserializationSchema}.
We would create a
@@ -35,6 +39,7 @@ import org.apache.pulsar.client.api.Message;
@Internal
public class PulsarTypeInformationWrapper<T> implements
PulsarDeserializationSchema<T> {
private static final long serialVersionUID = 6647084180084963022L;
+ private static final Logger LOG =
LoggerFactory.getLogger(PulsarTypeInformationWrapper.class);
/**
* PulsarDeserializationSchema would be shared for multiple SplitReaders
in different fetcher
@@ -47,9 +52,11 @@ public class PulsarTypeInformationWrapper<T> implements
PulsarDeserializationSch
private final TypeInformation<T> information;
private final TypeSerializer<T> serializer;
- public PulsarTypeInformationWrapper(TypeInformation<T> information,
ExecutionConfig config) {
+ public PulsarTypeInformationWrapper(
+ TypeInformation<T> information, @Nullable ExecutionConfig config) {
this.information = information;
- this.serializer = information.createSerializer(config);
+ final ExecutionConfig executionConfig = config == null ? new
ExecutionConfig() : config;
+ this.serializer =
information.createSerializer(executionConfig.getSerializerConfig());
}
@Override
@@ -57,7 +64,20 @@ public class PulsarTypeInformationWrapper<T> implements
PulsarDeserializationSch
DataInputDeserializer dis = DESERIALIZER.get();
dis.setBuffer(message.getData());
T instance = serializer.deserialize(dis);
-
+ MessageIdAdv messageIdAdv = (MessageIdAdv) message.getMessageId();
+ if (LOG.isDebugEnabled()) {
+ if (messageIdAdv != null) {
+ LOG.debug(
+ "Deserialize message {}:{}:{}/{} of {}",
+ messageIdAdv.getLedgerId(),
+ messageIdAdv.getEntryId(),
+ messageIdAdv.getBatchIndex(),
+ messageIdAdv.getBatchSize(),
+ String.valueOf(instance));
+ } else {
+ LOG.debug("Deserialize message {id-null} of {}",
String.valueOf(instance));
+ }
+ }
out.collect(instance);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchemaFactory.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchemaFactory.java
index 95c443b..cff2ca7 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchemaFactory.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchemaFactory.java
@@ -15,6 +15,7 @@
package org.apache.flink.connector.pulsar.table.sink;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils;
import
org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.connector.format.EncodingFormat;
@@ -22,7 +23,6 @@ import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.utils.DataTypeUtils;
import javax.annotation.Nullable;
@@ -107,7 +107,8 @@ public class PulsarTableSerializationSchemaFactory {
}
DataType physicalFormatDataType =
Projection.of(projection).project(this.physicalDataType);
if (prefix != null) {
- physicalFormatDataType =
DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
+ physicalFormatDataType =
+ PulsarSerdeUtils.stripRowPrefix(physicalFormatDataType,
prefix);
}
return format.createRuntimeEncoder(context, physicalFormatDataType);
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java
index 32294b8..91515a8 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.pulsar.table.source;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils;
import
org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.connector.format.DecodingFormat;
@@ -27,7 +28,6 @@ import
org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.utils.DataTypeUtils;
import javax.annotation.Nullable;
@@ -117,7 +117,8 @@ public class PulsarTableDeserializationSchemaFactory
implements Serializable {
DataType physicalFormatDataType =
Projection.of(projection).project(this.physicalDataType);
if (prefix != null) {
- physicalFormatDataType =
DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
+ physicalFormatDataType =
+ PulsarSerdeUtils.stripRowPrefix(physicalFormatDataType,
prefix);
}
return format.createRuntimeDecoder(context, physicalFormatDataType);
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java
index 83a7f36..af980fd 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java
@@ -25,6 +25,7 @@ import
org.apache.flink.connector.testframe.environment.TestEnvironment;
import
org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
import org.apache.flink.core.execution.JobClient;
import
org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.minicluster.RpcServiceSharing;
import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
import org.apache.flink.runtime.testutils.CommonTestUtils;
@@ -49,7 +50,6 @@ import java.util.stream.Collectors;
import static
org.apache.flink.configuration.MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL;
import static
org.apache.flink.connector.testframe.utils.ConnectorTestConstants.METRIC_FETCHER_UPDATE_INTERVAL_MS;
-import static
org.apache.flink.runtime.jobgraph.SavepointConfigOptions.SAVEPOINT_PATH;
/** Test environment for running jobs on Flink mini-cluster. */
@Experimental
@@ -91,7 +91,9 @@ public class MiniClusterTestEnvironment implements
TestEnvironment, ClusterContr
TestEnvironmentSettings envOptions) {
Configuration configuration = new Configuration();
if (envOptions.getSavepointRestorePath() != null) {
- configuration.setString(SAVEPOINT_PATH,
envOptions.getSavepointRestorePath());
+ SavepointRestoreSettings.toConfiguration(
+
SavepointRestoreSettings.forPath(envOptions.getSavepointRestorePath()),
+ configuration);
}
return new TestStreamEnvironment(
this.miniCluster.getMiniCluster(),
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
index 2f56f90..853d385 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
@@ -40,7 +40,7 @@ class PulsarSchemaTypeInformationTest {
PulsarSchemaTypeInformation<Bar> clonedInfo =
InstantiationUtil.clone(info);
assertThat(clonedInfo).isEqualTo(info).isNotSameAs(info);
- assertThatCode(() -> info.createSerializer(new ExecutionConfig()))
+ assertThatCode(() -> info.createSerializer(new
ExecutionConfig().getSerializerConfig()))
.doesNotThrowAnyException();
assertThat(clonedInfo.getTypeClass()).isEqualTo(info.getTypeClass());
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
index 14d907d..0729bcf 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
@@ -91,7 +91,7 @@ class AvroSchemaFactoryTest {
// Serialize by type information.
TypeSerializer<StructWithAnnotations> serializer =
- information.createSerializer(new ExecutionConfig());
+ information.createSerializer(new
ExecutionConfig().getSerializerConfig());
// TypeInformation serialization.
assertThatCode(() ->
InstantiationUtil.clone(information)).doesNotThrowAnyException();
assertThatCode(() ->
InstantiationUtil.clone(serializer)).doesNotThrowAnyException();
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
index d24f230..31ffd1b 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -20,13 +20,15 @@ package org.apache.flink.connector.pulsar.sink.writer;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.api.common.TaskInfo;
+import org.apache.flink.api.common.TaskInfoImpl;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
@@ -162,12 +164,14 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
}
}
- private static class MockInitContext implements InitContext {
+ private static class MockInitContext implements WriterInitContext {
private final MetricListener metricListener;
private final OperatorIOMetricGroup ioMetricGroup;
private final SinkWriterMetricGroup metricGroup;
private final ProcessingTimeService timeService;
+ private final JobInfo jobInfo;
+ private final TaskInfo taskInfo;
private MockInitContext() {
this.metricListener = new MetricListener();
@@ -178,6 +182,8 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
this.metricGroup =
MetricsGroupTestUtils.mockWriterMetricGroup(metricGroup,
ioMetricGroup);
this.timeService = new TestProcessingTimeService();
+ this.jobInfo = new JobInfoImpl(new JobID(), "pulsar-writer-test");
+ this.taskInfo = new TaskInfoImpl("pulsar-writer-test-task", 1, 0,
1, 0);
}
@Override
@@ -196,39 +202,15 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
}
@Override
- public int getSubtaskId() {
- return 0;
- }
-
- @Override
- public int getNumberOfParallelSubtasks() {
- return 1;
- }
-
- @Override
- public int getAttemptNumber() {
- return 0;
- }
-
- // The following three methods are for compatibility with
- //
https://github.com/apache/flink/commit/4f5b2fb5736f5a1c098a7dc1d448a879f36f801b
- // . Removed the commented out `@Override` when we move to 1.18.
-
- // @Override
public boolean isObjectReuseEnabled() {
return false;
}
- // @Override
+ @Override
public <IN> TypeSerializer<IN> createInputSerializer() {
return null;
}
- // @Override
- public JobID getJobId() {
- return null;
- }
-
@Override
public SinkWriterMetricGroup metricGroup() {
return metricGroup;
@@ -257,12 +239,12 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
@Override
public JobInfo getJobInfo() {
- return null;
+ return jobInfo;
}
@Override
public TaskInfo getTaskInfo() {
- return null;
+ return taskInfo;
}
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
index 9e0dabf..03fef2b 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
@@ -28,9 +28,9 @@ import
org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SchemaSerializationException;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
@@ -113,9 +113,11 @@ class ProducerRegisterTest extends PulsarTestSuiteBase {
sinkConfiguration, PulsarCrypto.disabled(),
createSinkWriterMetricGroup());
long message = ThreadLocalRandom.current().nextLong();
- TypedMessageBuilder<byte[]> builder =
register.createMessageBuilder(topic, Schema.BYTES);
-
- assertThatThrownBy(() -> builder.value(Schema.INT64.encode(message)))
+ TypedMessageBuilderImpl<Object> builder =
+ (TypedMessageBuilderImpl<Object>)
+ register.createMessageBuilder(topic, Schema.BYTES);
+ builder.value(Schema.INT64.encode(message));
+ assertThatThrownBy(() -> builder.getMessage())
.isInstanceOf(SchemaSerializationException.class);
}
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
index 63b2da7..8c427f8 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.source;
import org.apache.flink.connector.pulsar.common.MiniClusterTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.SimpleCollectIteratorAssert;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
import
org.apache.flink.connector.pulsar.testutils.source.cases.EncryptedMessagesConsumingContext;
import
org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
@@ -31,10 +32,16 @@ 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.connector.testframe.utils.CollectIteratorAssertions;
import org.apache.flink.core.execution.CheckpointingMode;
+import org.apache.flink.util.CloseableIterator;
import org.apache.pulsar.client.api.SubscriptionType;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
import static org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE;
/**
@@ -70,4 +77,34 @@ class PulsarSourceITCase extends SourceTestSuiteBase<String>
{
@TestContext
PulsarTestContextFactory<String, EncryptedMessagesConsumingContext>
encryptMessages =
new PulsarTestContextFactory<>(pulsar,
EncryptedMessagesConsumingContext::new);
+
+ /**
+ * {@link CollectIteratorAssertions} will generate a mismatch description
if the result does not
+ * match the expected value. It attempts to capture all following
messages, even though already
+ * failed, which helps engineers for troubleshooting, but draining the
following messages may
+ * lead the test to get stuck. We rewrite the method to avoid the test to
get stuck.
+ */
+ @Override
+ protected void checkResultWithSemantic(
+ CloseableIterator<String> resultIterator,
+ List<List<String>> expectedData,
+ org.apache.flink.core.execution.CheckpointingMode semantic,
+ Integer limit) {
+ if (limit != null) {
+ Runnable runnable =
+ () ->
+ new SimpleCollectIteratorAssert<>(resultIterator)
+ .withNumRecordsLimit(limit)
+ .matchesRecordsFromSource(expectedData,
semantic);
+
+ try {
+ CompletableFuture.runAsync(runnable).get(65, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new AssertionError("Timed out or failed while validating
source results.", e);
+ }
+ } else {
+ new SimpleCollectIteratorAssert<>(resultIterator)
+ .matchesRecordsFromSource(expectedData, semantic);
+ }
+ }
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReaderTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReaderTest.java
index 4b439a0..7fae15b 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReaderTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReaderTest.java
@@ -72,7 +72,7 @@ class PulsarPartitionSplitReaderTest extends
PulsarTestSuiteBase {
String topicName = randomAlphabetic(10);
// Add a split
- handleSplit(splitReader, topicName, 0, MessageId.latest);
+ handleSplit(splitReader, topicName, 0, MessageId.earliest);
// Poll once with a null message
Message<byte[]> message1 = fetchedMessage(splitReader);
@@ -97,7 +97,7 @@ class PulsarPartitionSplitReaderTest extends
PulsarTestSuiteBase {
PulsarPartitionSplitReader splitReader = splitReader();
String topicName = randomAlphabetic(10);
- handleSplit(splitReader, topicName, 0, MessageId.latest);
+ handleSplit(splitReader, topicName, 0, MessageId.earliest);
operator().sendMessage(topicNameWithPartition(topicName, 0), STRING,
randomAlphabetic(10));
fetchedMessages(splitReader, 1, true);
}
@@ -263,16 +263,20 @@ class PulsarPartitionSplitReaderTest extends
PulsarTestSuiteBase {
.admin()
.topics()
.getLastMessageId(topicNameWithPartition(topicName, 0));
- // when recover, use exclusive startCursor
- handleSplit(
- splitReader,
- topicName,
- 0,
+ MessageId startMessageId =
new MessageIdImpl(
lastMessageId.getLedgerId(),
lastMessageId.getEntryId() - 1,
- lastMessageId.getPartitionIndex()));
- fetchedMessages(splitReader, 1, true);
+ lastMessageId.getPartitionIndex());
+ try {
+ // when recover, use exclusive startCursor
+ handleSplit(splitReader, topicName, 0, startMessageId);
+ fetchedMessages(splitReader, 1, true);
+ } catch (Throwable t) {
+ throw t;
+ } finally {
+ splitReader.close();
+ }
}
@Test
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
index 3d4619d..eedf4b8 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
@@ -56,6 +56,7 @@ import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import static java.util.Collections.emptyList;
@@ -143,45 +144,56 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
void offsetCommitOnCheckpointComplete() throws Exception {
String topicName = topicName();
PulsarSourceReader<Integer> reader = sourceReader();
+ AtomicBoolean readerClosed = new AtomicBoolean(false);
// consume more than 1 partition
- reader.addSplits(
- createPartitionSplits(
- topicName, DEFAULT_PARTITIONS,
Boundedness.CONTINUOUS_UNBOUNDED));
- reader.notifyNoMoreSplits();
- TestingReaderOutput<Integer> output = new TestingReaderOutput<>();
- long checkpointId = 0;
- int emptyResultTime = 0;
- InputStatus status;
- do {
- checkpointId++;
- status = reader.pollNext(output);
- // Create a checkpoint for each message consumption, but not
complete them.
- reader.snapshotState(checkpointId);
- // the first couple of pollNext() might return NOTHING_AVAILABLE
before data appears
- if (InputStatus.NOTHING_AVAILABLE == status) {
- emptyResultTime++;
- sleepUninterruptibly(1, TimeUnit.SECONDS);
+ try {
+ reader.addSplits(
+ createPartitionSplits(
+ topicName, DEFAULT_PARTITIONS,
Boundedness.CONTINUOUS_UNBOUNDED));
+ reader.notifyNoMoreSplits();
+ TestingReaderOutput<Integer> output = new TestingReaderOutput<>();
+ long checkpointId = 0;
+ int emptyResultTime = 0;
+ InputStatus status;
+ do {
+ checkpointId++;
+ status = reader.pollNext(output);
+ // Create a checkpoint for each message consumption, but not
complete them.
+ reader.snapshotState(checkpointId);
+ // the first couple of pollNext() might return
NOTHING_AVAILABLE before data appears
+ if (InputStatus.NOTHING_AVAILABLE == status) {
+ emptyResultTime++;
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+
+ } while (emptyResultTime < MAX_EMPTY_POLLING_TIMES
+ && status != InputStatus.END_OF_INPUT
+ && output.getEmittedRecords().size()
+ < NUM_RECORDS_PER_PARTITION * DEFAULT_PARTITIONS);
+
+ // The completion of the last checkpoint should subsume all
previous checkpoints.
+ assertThat(reader.cursorsToCommit).hasSize((int) checkpointId);
+ long lastCheckpointId = checkpointId;
+ // notify checkpoint complete and expect all cursors committed
+ assertThatCode(() ->
reader.notifyCheckpointComplete(lastCheckpointId))
+ .doesNotThrowAnyException();
+ assertThat(reader.cursorsToCommit).isEmpty();
+
+ // Verify the committed offsets.
+ readerClosed.set(true);
+ reader.close();
+ for (int i = 0; i < DEFAULT_PARTITIONS; i++) {
+ verifyAllMessageAcknowledged(
+ NUM_RECORDS_PER_PARTITION,
+ TopicNameUtils.topicNameWithPartition(topicName, i));
+ }
+ } catch (Throwable t) {
+ throw t;
+ } finally {
+ if (!readerClosed.get()) {
+ reader.close();
}
-
- } while (emptyResultTime < MAX_EMPTY_POLLING_TIMES
- && status != InputStatus.END_OF_INPUT
- && output.getEmittedRecords().size()
- < NUM_RECORDS_PER_PARTITION * DEFAULT_PARTITIONS);
-
- // The completion of the last checkpoint should subsume all previous
checkpoints.
- assertThat(reader.cursorsToCommit).hasSize((int) checkpointId);
- long lastCheckpointId = checkpointId;
- // notify checkpoint complete and expect all cursors committed
- assertThatCode(() -> reader.notifyCheckpointComplete(lastCheckpointId))
- .doesNotThrowAnyException();
- assertThat(reader.cursorsToCommit).isEmpty();
-
- // Verify the committed offsets.
- reader.close();
- for (int i = 0; i < DEFAULT_PARTITIONS; i++) {
- verifyAllMessageAcknowledged(
- NUM_RECORDS_PER_PARTITION,
TopicNameUtils.topicNameWithPartition(topicName, i));
}
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java
index eb83967..fc0b9ea 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.connector.pulsar.table.testutils.TestingUser;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.test.util.SuccessException;
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableTestBase.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableTestBase.java
index cecd3e2..40eb78a 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableTestBase.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableTestBase.java
@@ -18,7 +18,8 @@
package org.apache.flink.connector.pulsar.table;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.connector.pulsar.common.MiniClusterTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
@@ -32,10 +33,14 @@ import
org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Base class for Pulsar table integration test. */
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public abstract class PulsarTableTestBase {
+ private static final Logger LOG =
LoggerFactory.getLogger(PulsarTableTestBase.class);
+
@TestEnv protected MiniClusterTestEnvironment flink = new
MiniClusterTestEnvironment();
// Defines pulsar running environment
@@ -54,11 +59,15 @@ public abstract class PulsarTableTestBase {
@BeforeAll
public void beforeAll() throws Exception {
+ LOG.info("Starting Pulsar table test environment for {}.",
getClass().getSimpleName());
pulsar.startUp();
+ LOG.info("Pulsar test environment started for {}.",
getClass().getSimpleName());
// run env
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(DEFAULT_PARALLELISM);
- env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ Configuration noRestart = new Configuration();
+ noRestart.set(RestartStrategyOptions.RESTART_STRATEGY, "disable");
+ env.configure(noRestart);
tableEnv = StreamTableEnvironment.create(env);
tableEnv.getConfig()
.getConfiguration()
@@ -71,6 +80,15 @@ public abstract class PulsarTableTestBase {
@AfterAll
public void afterAll() throws Exception {
- pulsar.tearDown();
+ final long start = System.currentTimeMillis();
+ LOG.info("Tearing down Pulsar table test environment for {}.",
getClass().getSimpleName());
+ try {
+ pulsar.tearDown();
+ } finally {
+ LOG.info(
+ "Finished Pulsar table test environment teardown for {} in
{} ms.",
+ getClass().getSimpleName(),
+ System.currentTimeMillis() - start);
+ }
}
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/SimpleCollectIteratorAssert.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/SimpleCollectIteratorAssert.java
new file mode 100644
index 0000000..13c445e
--- /dev/null
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/SimpleCollectIteratorAssert.java
@@ -0,0 +1,279 @@
+/*
+ * 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;
+
+import org.apache.flink.core.execution.CheckpointingMode;
+
+import org.assertj.core.api.AbstractAssert;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+import static
org.apache.flink.streaming.api.CheckpointingMode.convertToCheckpointingMode;
+
+/**
+ * Difference with {@link
org.apache.flink.connector.testframe.utils.CollectIteratorAssertions}
+ * CollectIteratorAssertions will generate a mismatch description if the
result does not match the
+ * expected value. It attempts to capture all following messages, even though
already failed, which
+ * helps engineers for troubleshooting, but draining the following messages
may lead the test to get
+ * stuck.
+ *
+ * <p>The current implementation will immediately fail once a message does not
match, skips to drain
+ * the following messages to build the information that describes failure.
+ */
+public class SimpleCollectIteratorAssert<T>
+ extends AbstractAssert<SimpleCollectIteratorAssert<T>, Iterator<T>> {
+
+ private final Iterator<T> collectorIterator;
+ private final List<RecordsFromSplit<T>> expectedRecordsFromSplits = new
ArrayList<>();
+ private int totalNumRecords;
+ private Integer limit = null;
+
+ public SimpleCollectIteratorAssert(Iterator<T> collectorIterator) {
+ super(collectorIterator, SimpleCollectIteratorAssert.class);
+ this.collectorIterator = collectorIterator;
+ }
+
+ public SimpleCollectIteratorAssert<T> withNumRecordsLimit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ /**
+ * This method is required for downstream projects e.g. Flink connectors
extending this test for
+ * the case when there should be supported Flink versions below 1.20.
Could be removed together
+ * with dropping support for Flink 1.19.
+ */
+ @Deprecated
+ public void matchesRecordsFromSource(
+ List<List<T>> recordsBySplitsFromSource,
+ org.apache.flink.streaming.api.CheckpointingMode semantic) {
+ matchesRecordsFromSource(recordsBySplitsFromSource,
convertToCheckpointingMode(semantic));
+ }
+
+ public void matchesRecordsFromSource(List<List<T>> expectedData,
CheckpointingMode semantic) {
+ for (List<T> recordsFromSplit : expectedData) {
+ expectedRecordsFromSplits.add(new
RecordsFromSplit<>(recordsFromSplit));
+ totalNumRecords += recordsFromSplit.size();
+ }
+
+ if (limit != null && limit > totalNumRecords) {
+ throw new IllegalArgumentException(
+ "Limit validation size should be less than total number of
records from source");
+ }
+
+ switch (semantic) {
+ case AT_LEAST_ONCE:
+ compareWithAtLeastOnceSemantic(collectorIterator,
expectedRecordsFromSplits);
+ break;
+ case EXACTLY_ONCE:
+ compareWithExactlyOnceSemantic(collectorIterator,
expectedRecordsFromSplits);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ String.format("Unrecognized semantic \"%s\"",
semantic));
+ }
+ }
+
+ private void compareWithAtLeastOnceSemantic(
+ Iterator<T> resultIterator, List<RecordsFromSplit<T>>
recordsFromSplits) {
+ List<T> duplicateRead = new LinkedList<>();
+
+ int recordCounter = 0;
+ while (resultIterator.hasNext()) {
+ final T record = resultIterator.next();
+ if (!matchThenNext(record)) {
+ duplicateRead.add(record);
+ } else {
+ recordCounter++;
+ }
+
+ if (limit != null && recordCounter >= limit) {
+ break;
+ }
+ }
+ if (limit == null && !hasReachedEnd()) {
+ failWithMessage(
+ generateMismatchDescription(
+ String.format(
+ "Expected to have at least %d records in
result, but only received %d records",
+ recordsFromSplits.stream()
+ .mapToInt(
+ recordsFromSplit ->
+
recordsFromSplit.records.size())
+ .sum(),
+ recordCounter),
+ resultIterator));
+ } else {
+ confirmDuplicateRead(duplicateRead);
+ }
+ }
+
+ private void compareWithExactlyOnceSemantic(
+ Iterator<T> resultIterator, List<RecordsFromSplit<T>>
expectedData) {
+ int recordCounter = 0;
+ while (resultIterator.hasNext()) {
+ final T recordReceived = resultIterator.next();
+ if (!matchThenNext(recordReceived)) {
+ if (recordCounter >= totalNumRecords) {
+ failWithMessage(
+ generateMismatchDescription(
+ String.format(
+ "Expected to have exactly %d
records in result, but received more records",
+ expectedData.stream()
+ .mapToInt(
+ recordsFromSplit ->
+
recordsFromSplit.records.size())
+ .sum()),
+ resultIterator));
+ } else {
+ failWithMessage(
+ generateMismatchDescription(
+ String.format(
+ "Unexpected record '%s' at
position %d(count of matched records)",
+ recordReceived, recordCounter),
+ resultIterator));
+ }
+ }
+ recordCounter++;
+ if (limit != null && recordCounter >= limit) {
+ break;
+ }
+ }
+ if (limit == null && !hasReachedEnd()) {
+ failWithMessage(
+ generateMismatchDescription(
+ String.format(
+ "Expected to have exactly %d records in
result, but only received %d records",
+ expectedData.stream()
+ .mapToInt(
+ recordsFromSplit ->
+
recordsFromSplit.records.size())
+ .sum(),
+ recordCounter),
+ resultIterator));
+ }
+ }
+
+ private void confirmDuplicateRead(List<T> duplicateRead) {
+ for (T record : duplicateRead) {
+ boolean found = false;
+ for (RecordsFromSplit<T> recordsFromSplit :
expectedRecordsFromSplits) {
+ if (recordsFromSplit.records.contains(record)) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ failWithMessage(String.format("Unexpected duplicate record
'%s'", record));
+ }
+ }
+ }
+
+ /**
+ * Check if any pointing data is identical to the record from the stream,
and move the pointer
+ * to next record if matched.
+ *
+ * @param record Record from stream
+ */
+ private boolean matchThenNext(T record) {
+ for (RecordsFromSplit<T> recordsFromSplit : expectedRecordsFromSplits)
{
+ if (!recordsFromSplit.hasNext()) {
+ continue;
+ }
+ if (record.equals(recordsFromSplit.current())) {
+ recordsFromSplit.forward();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether all pointers have reached the end of lists.
+ *
+ * @return True if all pointers have reached the end.
+ */
+ private boolean hasReachedEnd() {
+ for (RecordsFromSplit<T> recordsFromSplit : expectedRecordsFromSplits)
{
+ if (recordsFromSplit.hasNext()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private String generateMismatchDescription(String reason, Iterator<T>
resultIterator) {
+ final StringBuilder sb = new StringBuilder();
+ sb.append(reason).append("\n");
+ sb.append("Current progress of multiple split test data
validation:\n");
+ int splitCounter = 0;
+ for (RecordsFromSplit<T> recordsFromSplit : expectedRecordsFromSplits)
{
+ sb.append(
+ String.format(
+ "Split %d (%d/%d): \n",
+ splitCounter++,
+ recordsFromSplit.offset,
+ recordsFromSplit.records.size()));
+ for (int recordIndex = 0;
+ recordIndex < recordsFromSplit.records.size();
+ recordIndex++) {
+ sb.append(recordsFromSplit.records.get(recordIndex));
+ if (recordIndex == recordsFromSplit.offset) {
+ sb.append("\t<---- failed to compare with the current
record");
+ }
+ sb.append("\n");
+ }
+ }
+ // Modify the following info building to avoid the test being stuck.
+ if (resultIterator.hasNext()) {
+ sb.append("Remaining received elements after the unexpected one:
xxx\n");
+ // while (resultIterator.hasNext()) {
+ // sb.append(resultIterator.next()).append("\n");
+ // }
+ }
+ return sb.toString();
+ }
+
+ private static class RecordsFromSplit<T> {
+ private int offset = 0;
+ private final List<T> records;
+
+ public RecordsFromSplit(List<T> records) {
+ this.records = records;
+ }
+
+ public T current() {
+ if (!hasNext()) {
+ return null;
+ }
+ return records.get(offset);
+ }
+
+ public void forward() {
+ ++offset;
+ }
+
+ public boolean hasNext() {
+ return offset < records.size();
+ }
+ }
+}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java
index 63db956..fb9211f 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java
@@ -25,7 +25,7 @@ import
org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.testutils.junit.SharedObjectsExtension;
import org.apache.flink.testutils.junit.SharedReference;
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
index 9bd2252..2b93e79 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
@@ -54,7 +54,7 @@ public class PulsarContainerRuntime implements PulsarRuntime {
private static final String PULSAR_ADMIN_URL =
String.format("http://%s:%d";, PULSAR_INTERNAL_HOSTNAME,
BROKER_HTTP_PORT);
- private static final String CURRENT_VERSION = "3.0.5";
+ private static final String CURRENT_VERSION = "4.0.13";
private final PulsarContainer container;
private final AtomicBoolean started;
diff --git a/flink-sql-connector-pulsar/pom.xml
b/flink-sql-connector-pulsar/pom.xml
index 2c35bf0..3629707 100644
--- a/flink-sql-connector-pulsar/pom.xml
+++ b/flink-sql-connector-pulsar/pom.xml
@@ -76,10 +76,13 @@ under the License.
<include>org.apache.pulsar:pulsar-client-admin-api</include>
<include>org.apache.pulsar:pulsar-client-all</include>
<include>org.apache.pulsar:pulsar-client-api</include>
-
<include>org.bouncycastle:bcpkix-jdk15on</include>
-
<include>org.bouncycastle:bcprov-ext-jdk15on</include>
-
<include>org.bouncycastle:bcprov-jdk15on</include>
-
<include>org.bouncycastle:bcutil-jdk15on</include>
+
<include>org.bouncycastle:bcpkix-jdk18on</include>
+
<include>org.bouncycastle:bcprov-jdk18on</include>
+
<include>org.bouncycastle:bcutil-jdk18on</include>
+
<include>io.opentelemetry:opentelemetry-api</include>
+
<include>io.opentelemetry:opentelemetry-api-incubator</include>
+
<include>io.opentelemetry:opentelemetry-context</include>
+
<include>io.opentelemetry:opentelemetry-common</include>
</includes>
</artifactSet>
<filters>
diff --git a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
index a36feb1..e9bc18f 100644
--- a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
+++ b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
@@ -6,15 +6,18 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- com.fasterxml.jackson.core:jackson-annotations:2.13.4
-- org.apache.pulsar:pulsar-client-admin-api:3.0.5
-- org.apache.pulsar:pulsar-client-all:3.0.5
-- org.apache.pulsar:pulsar-client-api:3.0.5
+- com.fasterxml.jackson.core:jackson-annotations:2.22
+- io.opentelemetry:opentelemetry-api:1.62.0
+- io.opentelemetry:opentelemetry-api-incubator:1.62.0-alpha
+- io.opentelemetry:opentelemetry-common:1.62.0
+- io.opentelemetry:opentelemetry-context:1.62.0
+- org.apache.pulsar:pulsar-client-admin-api:4.0.13
+- org.apache.pulsar:pulsar-client-all:4.0.13
+- org.apache.pulsar:pulsar-client-api:4.0.13
This project bundles the following dependencies under the Bouncy Castle
license.
See bundled license files for details.
-- org.bouncycastle:bcpkix-jdk15on:1.69
-- org.bouncycastle:bcprov-ext-jdk15on:1.69
-- org.bouncycastle:bcprov-jdk15on:1.69
-- org.bouncycastle:bcutil-jdk15on:1.69
+- org.bouncycastle:bcpkix-jdk18on:1.84
+- org.bouncycastle:bcprov-jdk18on:1.84
+- org.bouncycastle:bcutil-jdk18on:1.84
diff --git a/pom.xml b/pom.xml
index 685d8ec..5c98841 100644
--- a/pom.xml
+++ b/pom.xml
@@ -51,11 +51,10 @@ under the License.
</modules>
<properties>
- <flink.version>1.20.3</flink.version>
- <pulsar.version>3.0.5</pulsar.version>
+ <flink.version>2.0.2</flink.version>
+ <pulsar.version>4.0.13</pulsar.version>
+ <opentelemetry.version>1.62.0</opentelemetry.version>
<scala.binary.version>2.12</scala.binary.version>
- <bouncycastle.version>1.69</bouncycastle.version>
-
<scala-reflect.version>2.12.7</scala-reflect.version>
<scala-library.version>2.12.7</scala-library.version>
@@ -64,7 +63,7 @@ under the License.
<assertj.version>3.23.1</assertj.version>
<mockito.version>4.11.0</mockito.version>
<archunit.version>1.0.1</archunit.version>
- <testcontainers.version>1.21.4</testcontainers.version>
+ <testcontainers.version>2.0.4</testcontainers.version>
<japicmp.skip>false</japicmp.skip>
<japicmp.referenceVersion>3.0.0-1.16</japicmp.referenceVersion>
@@ -76,10 +75,13 @@ under the License.
<commons-compress.version>1.26.1</commons-compress.version>
<commons-lang3.version>3.14.0</commons-lang3.version>
<commons-io.version>2.15.1</commons-io.version>
+ <caffeine.version>2.9.1</caffeine.version>
+ <checker-qual.version>3.12.0</checker-qual.version>
+ <errorprone.version>2.11.0</errorprone.version>
<byte-buddy.version>1.12.20</byte-buddy.version>
<kryo.version>2.24.0</kryo.version>
<objenesis.version>3.3</objenesis.version>
- <jackson-bom.version>2.13.4.20221013</jackson-bom.version>
+ <jackson-bom.version>2.22.2</jackson-bom.version>
<snappy.java.version>1.1.10.4</snappy.java.version>
<os-maven-plugin.version>1.7.0</os-maven-plugin.version>
@@ -134,7 +136,7 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
- <artifactId>junit-jupiter</artifactId>
+ <artifactId>testcontainers-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
@@ -391,6 +393,18 @@ under the License.
<version>${jackson-bom.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api</artifactId>
+ <version>${opentelemetry.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api-incubator</artifactId>
+ <version>${opentelemetry.version}-alpha</version>
+ </dependency>
+
<!-- Start of pinned versions for dependency convergence -->
<dependency>
@@ -411,6 +425,24 @@ under the License.
<version>${commons-io.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ <version>${caffeine.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.checkerframework</groupId>
+ <artifactId>checker-qual</artifactId>
+ <version>${checker-qual.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.errorprone</groupId>
+ <artifactId>error_prone_annotations</artifactId>
+ <version>${errorprone.version}</version>
+ </dependency>
+
<dependency>
<groupId>net.bytebuddy</groupId>
<artifactId>byte-buddy</artifactId>
