Skip to content

KAFKA-12499: add transaction timeout verification #10482

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
abbccdda merged 4 commits into from
May 21, 2021
Merged

KAFKA-12499: add transaction timeout verification #10482

abbccdda merged 4 commits into from
May 21, 2021

Conversation

@abbccdda
Copy link
Contributor

@abbccdda abbccdda commented Apr 6, 2021

This PR tries to add the check for transaction timeout for a comparison against commit interval of streams. If transaction timeout is smaller than commit interval, stream should crash and inform user to update their commit interval to match the given transaction timeout, or vise versa.

Committer Checklist (excluded from commit message)

  • Verify design and implementation
  • Verify test coverage and CI build status
  • Verify documentation (including upgrade notes)

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch from 58983d0 to 067e5dc Compare April 6, 2021 00:16
mjsax
mjsax reviewed Apr 6, 2021
View reviewed changes
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java Outdated
Copy link
Member

@mjsax mjsax Apr 6, 2021

Choose a reason for hiding this comment

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

It might be better to do this check in StreamsConfig ?

Copy link
Member

@ableegoldman ableegoldman Apr 6, 2021

Choose a reason for hiding this comment

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

+1, we should fail-fast at the level of the KafkaStreams. Otherwise in eos-alpha we actually have to wait for all the individual StreamThreads to be created, started, complete a rebalance to get a task assignment, and then start throwing IllegalArgumentException on each thread when it goes to create a task producer. (Can we also log an error in addition to the IllegalArgumentException?)

Copy link
Contributor Author

@abbccdda abbccdda Apr 6, 2021

Choose a reason for hiding this comment

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

Sounds good!

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch 2 times, most recently from 8974909 to 961c2fb Compare April 6, 2021 23:15
guozhangwang
guozhangwang approved these changes Apr 7, 2021
View reviewed changes
Copy link
Contributor

@guozhangwang guozhangwang left a comment

Choose a reason for hiding this comment

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

LGTM!

@guozhangwang
Copy link
Contributor

guozhangwang commented Apr 7, 2021

Re-triggered unit tests

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch 2 times, most recently from 66ba7c2 to 9a87b6a Compare April 7, 2021 20:45
mjsax
mjsax reviewed Apr 8, 2021
View reviewed changes
Copy link
Member

@mjsax mjsax left a comment

Choose a reason for hiding this comment

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

One more comment.

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java Outdated
Copy link
Member

@mjsax mjsax Apr 8, 2021

Choose a reason for hiding this comment

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

We should use long as commitInterval is declared a LONG in StreamsConfig:

.define(COMMIT_INTERVAL_MS_CONFIG,
                    Type.LONG,
                    DEFAULT_COMMIT_INTERVAL_MS,
                    atLeast(0),
                    Importance.LOW,
                    COMMIT_INTERVAL_MS_DOC)

Copy link
Contributor Author

@abbccdda abbccdda Apr 8, 2021

Choose a reason for hiding this comment

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

We need to either convert transaction timeout or commit interval I assume. Will switch to long comparison instead.

Copy link
Member

@mjsax mjsax Apr 9, 2021

Choose a reason for hiding this comment

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

That makes sense -- a long -> int cast is safe, but not the other way around.

...ams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewTypedDemo.java Outdated
Copy link
Member

@mjsax mjsax Apr 8, 2021

Choose a reason for hiding this comment

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

Why do we update all this? While not incorrect, should not be necessary.

Copy link
Contributor Author

@abbccdda abbccdda Apr 8, 2021

Choose a reason for hiding this comment

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

Just realized that it's safer to use the correct long type instead of having int/long/string mixed as much as possible.

@ableegoldman
Copy link
Member

ableegoldman commented Apr 8, 2021

Build failed with Execution failed for task ':streams:checkstyleMain

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch 2 times, most recently from 7475fdb to 7b84ad4 Compare April 9, 2021 00:57
mjsax
mjsax reviewed Apr 9, 2021
View reviewed changes
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java Outdated
Copy link
Member

@mjsax mjsax Apr 9, 2021

Choose a reason for hiding this comment

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

In the explicit cast required?

Copy link
Contributor Author

@abbccdda abbccdda Apr 9, 2021

Choose a reason for hiding this comment

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

Yea, I think so.

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java Outdated
Copy link
Member

@mjsax mjsax Apr 9, 2021

Choose a reason for hiding this comment

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

It might be simpler to just use int transactionTimeout -- Java will auto-cast to long in

if (transactionTimeout < commitInterval) {

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch 2 times, most recently from b4e2d74 to 6bfa1c8 Compare April 9, 2021 23:21
mjsax
mjsax approved these changes Apr 10, 2021
View reviewed changes
Copy link
Member

@mjsax mjsax left a comment

Choose a reason for hiding this comment

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

Sorry for being picky... LGTM.

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch from 6bfa1c8 to 7050bef Compare April 10, 2021 21:03
@ableegoldman
Copy link
Member

ableegoldman commented Apr 23, 2021

@abbccdda it looks like some eos test failed in the PR build, guessing it's related

@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch from 7050bef to c01bd12 Compare May 19, 2021 02:05
@abbccdda abbccdda force-pushed the commit-interval-against-txn-timeout-KAFKA-12499 branch from 61d43d3 to 0e7b264 Compare May 19, 2021 06:39
@abbccdda
Copy link
Contributor Author

abbccdda commented May 20, 2021

Only unrelated test fails, merging.

@abbccdda abbccdda merged commit ae8b784 into apache:trunk May 21, 2021
ijuma added a commit to ijuma/kafka that referenced this pull request May 26, 2021
@ijuma
Merge remote-tracking branch 'apache-github/trunk' into reduce-produc…
9daeb3c 
…e-allocations-lz4

* apache-github/trunk: (43 commits)
  KAFKA-12800: Configure generator to fail on trailing JSON tokens (apache#10717)
  MINOR: clarify message ordering with max in-flight requests and idempotent producer (apache#10690)
  MINOR: Add log identifier/prefix printing in Log layer static functions (apache#10742)
  MINOR: update java doc for deprecated methods (apache#10722)
  MINOR: Fix deprecation warnings in SlidingWindowedCogroupedKStreamImplTest (apache#10703)
  KAFKA-12499: add transaction timeout verification (apache#10482)
  KAFKA-12620 Allocate producer ids on the controller (apache#10504)
  MINOR: Kafka Streams code samples formating unification (apache#10651)
  KAFKA-12808: Remove Deprecated Methods under StreamsMetrics (apache#10724)
  KAFKA-12522: Cast SMT should allow null value records to pass through (apache#10375)
  KAFKA-12820: Upgrade maven-artifact dependency to resolve CVE-2021-26291
  HOTFIX: fix checkstyle issue in KAFKA-12697
  KAFKA-12697: Add OfflinePartitionCount and PreferredReplicaImbalanceCount metrics to Quorum Controller (apache#10572)
  KAFKA-12342: Remove MetaLogShim and use RaftClient directly (apache#10705)
  KAFKA-12779: KIP-740, Clean up public API in TaskId and fix TaskMetadata#taskId() (apache#10735)
  KAFKA-12814: Remove Deprecated Method StreamsConfig getConsumerConfigs (apache#10737)
  MINOR: Eliminate redundant functions in LogTest suite (apache#10732)
  MINOR: Remove unused maxProducerIdExpirationMs parameter in Log constructor (apache#10723)
  MINOR: Updating files with release 2.7.1 (apache#10660)
  KAFKA-12809: Remove deprecated methods of Stores factory (apache#10729)
  ...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@mjsax mjsax mjsax approved these changes

@guozhangwang guozhangwang guozhangwang approved these changes

@ableegoldman ableegoldman Awaiting requested review from ableegoldman

Assignees

No one assigned

Labels

streams

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

4 participants

@abbccdda @guozhangwang @ableegoldman @mjsax