Skip to content
This repository has been archived by the owner on Jan 24, 2024. It is now read-only.

[transaction] Use a dedicated timeout for writing transaction marker #1968

Closed
wants to merge 1 commit into from
Closed

[transaction] Use a dedicated timeout for writing transaction marker #1968

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,12 @@
@Getter
public class KafkaRequestHandler extends KafkaCommandDecoder {
private static final int THROTTLE_TIME_MS = 10;
/**
* Request timeout for writes of the TXMARKERS
* Writing the TXMARKERS require recovery of the
* transactions on the PartitionLog at it may take much time.
*/
private static final int WRITE_TXN_MARKERS_TIMEOUT = 120000;
Copy link
Member

Choose a reason for hiding this comment

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

Why do we need to use this dedicated timeout?

Copy link
Contributor

Choose a reason for hiding this comment

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

This was only because we had problems during failure injection tests and we had to troubleshot.
it is a left over

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Thanks, I'll close this PR first.

private static final String POLICY_ROOT = "/admin/policies/";

private final PulsarService pulsarService;
Expand Down Expand Up @@ -2441,7 +2447,7 @@ protected void handleWriteTxnMarkers(KafkaHeaderAndRequest kafkaHeaderAndRequest
this.pendingTopicFuturesMap,
ctx);
getReplicaManager().appendRecords(
kafkaConfig.getRequestTimeoutMs(),
WRITE_TXN_MARKERS_TIMEOUT,
(short) 1,
true,
currentNamespacePrefix(),
Expand Down
Loading