Skip to content

[fix][io] Kafka Source connector maybe stuck #22511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
shibd merged 2 commits into from
Apr 16, 2024
Merged

[fix][io] Kafka Source connector maybe stuck #22511

shibd merged 2 commits into from
Apr 16, 2024

Conversation

@shibd
Copy link
Member

@shibd shibd commented Apr 15, 2024 •
edited
Loading

Motivation

The current implementation of Kafka source connector, that KafkaRecord does not implement the fail() method.

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java

Line 222 in bbae607

protected static class KafkaRecord<V> implements Record<V> {

If PulsarSink send a message to pulsar failed, will call record.fail().

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

Lines 194 to 196 in 43a9898

if (failSource) {
srcRecord.fail();
}

But because KafkaRecord does not implement it, this futures will never end.

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java

Line 178 in 495b141

CompletableFuture.allOf(futures).get();

This causes the Kafka consumer to leave the consumer group due to more than max.poll.interval.ms.

2024-04-11T14:06:25,730-0400 [kafka-coordinator-heartbeat-thread | kafka2pulsarConsumer] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer-g-1] Member consumer-1 sending LeaveGroup request to coordinator localhost:9092 (id: 2147483644 rack: null) due to consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

Modifications

  • Implement fail() method for KakfaRecord.
  • Let CompletableFuture.allOf(futures).get() support a timeout, set to 2/3 of max.poll.interval.ms. This adjustment is to ensure the interval between two poll requests does not exceed Kafka max.poll.interval.ms.
    • When it times out, it will trigger a connector failure, thereby restarting the connector.

Verifying this change

  • Add throwExceptionBySendFail and throwExceptionBySendTimeOut to cover it.

Documentation

  • doc
  • doc-required
  • doc-not-needed
  • doc-complete

Matching PR in forked repository

PR in forked repository:

@shibd shibd self-assigned this Apr 15, 2024
@github-actions
Copy link

github-actions bot commented Apr 15, 2024

@shibd Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

@github-actions github-actions bot added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Apr 15, 2024
lhotari
lhotari approved these changes Apr 15, 2024
View reviewed changes
Copy link
Member

@lhotari lhotari left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM. Good work @shibd

@lhotari lhotari added release/3.1.4 release/3.2.3 release/3.0.5 type/bug The PR fixed a bug or issue reported a bug labels Apr 15, 2024
Demogorgon314
Demogorgon314 approved these changes Apr 15, 2024
View reviewed changes
Copy link
Member

@Demogorgon314 Demogorgon314 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

nicoloboschi
nicoloboschi approved these changes Apr 15, 2024
View reviewed changes
Copy link
Contributor

@nicoloboschi nicoloboschi left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@shibd shibd merged commit bbff29d into apache:master Apr 16, 2024
shibd added a commit that referenced this pull request Apr 16, 2024
@shibd
[fix][io] Kafka Source connector maybe stuck (#22511)
1eafb06 
(cherry picked from commit bbff29d)
shibd added a commit that referenced this pull request Apr 16, 2024
@shibd
[fix][io] Kafka Source connector maybe stuck (#22511)
a0120d0 
(cherry picked from commit bbff29d)
shibd added a commit that referenced this pull request Apr 16, 2024
@shibd
[fix][io] Kafka Source connector maybe stuck (#22511)
3a2fb4e 
(cherry picked from commit bbff29d)
shibd added a commit that referenced this pull request Apr 16, 2024
@shibd
[fix][io] Kafka Source connector maybe stuck (#22511)
df6a182 
(cherry picked from commit bbff29d)
mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 16, 2024
@shibd @mukesh-ctds
[fix][io] Kafka Source connector maybe stuck (apache#22511)
8c5ad45 
(cherry picked from commit bbff29d)
(cherry picked from commit a0120d0)
@Technoboy- Technoboy- added this to the 3.3.0 milestone Apr 17, 2024
mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 17, 2024
@shibd @mukesh-ctds
[fix][io] Kafka Source connector maybe stuck (apache#22511)
73310b6 
(cherry picked from commit bbff29d)
(cherry picked from commit a0120d0)
mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 17, 2024
@shibd @mukesh-ctds
[fix][io] Kafka Source connector maybe stuck (apache#22511)
7b5858c 
(cherry picked from commit bbff29d)
(cherry picked from commit a0120d0)
mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 19, 2024
@shibd @mukesh-ctds
[fix][io] Kafka Source connector maybe stuck (apache#22511)
c8190ad 
(cherry picked from commit bbff29d)
(cherry picked from commit a0120d0)
srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 23, 2024
@shibd @srinath-ctds
[fix][io] Kafka Source connector maybe stuck (apache#22511)
c87fa4e 
(cherry picked from commit bbff29d)
(cherry picked from commit a0120d0)
hanmz pushed a commit to hanmz/pulsar that referenced this pull request Feb 12, 2025
@shibd @hanmz
[fix][io] Kafka Source connector maybe stuck (apache#22511)
bfe9635
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@lhotari lhotari lhotari approved these changes

@Demogorgon314 Demogorgon314 Demogorgon314 approved these changes

@nicoloboschi nicoloboschi nicoloboschi approved these changes

@Technoboy- Technoboy- Awaiting requested review from Technoboy-

@RobertIndie RobertIndie Awaiting requested review from RobertIndie

@poorbarcode poorbarcode Awaiting requested review from poorbarcode

Assignees

@shibd shibd

Labels

cherry-picked/branch-2.11 cherry-picked/branch-3.0 cherry-picked/branch-3.1 cherry-picked/branch-3.2 doc-not-needed Your PR changes do not impact docs ready-to-test release/2.11.5 release/3.0.5 release/3.1.4 release/3.2.3 type/bug The PR fixed a bug or issue reported a bug

Projects

None yet

Milestone

3.3.0

Development

Successfully merging this pull request may close these issues.

5 participants

@shibd @lhotari @Demogorgon314 @nicoloboschi @Technoboy-