cherry-pick fix: pulsar error during recovery (#17112) (#17129)

lmatz · NiuBlibing · web-flow · commit 4cda660772cf · 2024-06-05T09:35:45.000Z
Co-authored-by: Cherilyn Buren &lt;88433283+NiuBlibing@users.noreply.github.com&gt;
diff --git a/src/connector/src/source/pulsar/source/reader.rs b/src/connector/src/source/pulsar/source/reader.rs
@@ -180,12 +180,16 @@ impl SplitReader for PulsarBrokerReader {
                     )
                 } else {
                     builder.with_options(
-                        ConsumerOptions::default().with_initial_position(InitialPosition::Earliest),
+                        ConsumerOptions::default()
+                            .with_initial_position(InitialPosition::Earliest)
+                            .durable(false),
                     )
                 }
             }
             PulsarEnumeratorOffset::Latest => builder.with_options(
-                ConsumerOptions::default().with_initial_position(InitialPosition::Latest),
+                ConsumerOptions::default()
+                    .with_initial_position(InitialPosition::Latest)
+                    .durable(false),
             ),
             PulsarEnumeratorOffset::MessageId(m) => {
                 if topic.starts_with("non-persistent://") {

Original file line number	Diff line number	Diff line change
`@@ -180,12 +180,16 @@ impl SplitReader for PulsarBrokerReader {`
`180`	`180`	`)`
`181`	`181`	`} else {`
`182`	`182`	`builder.with_options(`
`183`		`- ConsumerOptions::default().with_initial_position(InitialPosition::Earliest),`
	`183`	`+ ConsumerOptions::default()`
	`184`	`+ .with_initial_position(InitialPosition::Earliest)`
	`185`	`+ .durable(false),`
`184`	`186`	`)`
`185`	`187`	`}`
`186`	`188`	`}`
`187`	`189`	`PulsarEnumeratorOffset::Latest => builder.with_options(`
`188`		`- ConsumerOptions::default().with_initial_position(InitialPosition::Latest),`
	`190`	`+ ConsumerOptions::default()`
	`191`	`+ .with_initial_position(InitialPosition::Latest)`
	`192`	`+ .durable(false),`
`189`	`193`	`),`
`190`	`194`	`PulsarEnumeratorOffset::MessageId(m) => {`
`191`	`195`	`if topic.starts_with("non-persistent://") {`