KAFKA-10555: Improve client state machine by wcarlson5 · Pull Request #9720 · apache/kafka

wcarlson5 · 2020-12-09T20:52:06Z

Removes the transition to error for when there are no threads and makes Error terminal

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

wcarlson5

@mjsax Can you give this a look? I hope that I haven't broken any of the EOS tests

wcarlson5 · 2020-12-10T20:44:14Z

This will not be needed with the new error definition

wcarlson5 · 2020-12-10T20:45:13Z

we will be doing the same thing but closing the client for now. Maybe a replace globalThread will be added later

wcarlson5 · 2020-12-10T20:48:17Z

We want to make close() idempotent and not throw an exception but we will log a warning, but only for close so that is why these logs are not in the setState() method.

wcarlson5 · 2020-12-10T20:48:50Z

This will let Streams shutdown uncleanly when in EOS mode

wcarlson5 · 2020-12-10T20:49:10Z

This test is for the functionality we are removing

wcarlson5 · 2020-12-10T23:36:11Z

Just wait for ERROR because you don't close crashed

wcarlson5 · 2020-12-10T23:36:33Z

No longer do you need to close crashed

Why this change? client2 should transit from PENDING_ERROR to ERROR (not from PENDING_SHUTDOWN to CLOSED, and thus the new CLOSE check seem not to be right? Or do I miss something?

My thoughts where that we are calling close and so its is going to just be closed as the state transitions are cleared. This is the test that made me want you to review the PR I was worried that I might have broken the integrity of the tests as it is not behaving as I expected. However no other test was having this problem so I was not sure if I was understanding it correctly.

From my understanding/ according to the new state transition diagram, if there is an error we go to PENDING_ERROR and if we call close() we transit PENDING_ERROR -> ERROR?

the call to close() will have no effect. The PENDING_ERROR -> ERROR transition is like the PENDING_SHUTDOWN -> NOT_RUNNING where the transition from the PENDING... state is automatic

Thanks for clarifying.

wcarlson5 · 2020-12-10T23:37:15Z

remove deprecated handler

wcarlson5 · 2020-12-10T23:41:41Z

When the client is shutdown it now goes to ERROR

mjsax · 2020-12-22T23:03:09Z

super nit: fix indention of comment

Also please update the state diagram in the comment above.

yay ascii art! good catch I forgot about that

When I build the docs locally it looks about right

mjsax · 2020-12-22T23:29:34Z

Should we return true there? I understand that we are not in NOT_RUNNING state, but in the end we are in a terminal state and we did cleanup all resources. -- I guess, I am raising the question if we should relax the definition of this return value?

I don't think that we should return true. A user would then expect that that the state would then change to NOT_RUNNING and could be stuck waiting on that. Where if we return false in then they might retry and get stuck there.

I don't know which is a better problem to have but I think that changing the meaning of this return value won't add any more clarity to the situation.

Fair point. Might be good to get input from somebody else in addition. \cc @cadonna

As far as I can see, where the return value is used the javadoc says

true if all threads were successfully stopped, false if the timeout was reached.

Since all threads were successfully stopped, I would return true. We clearly document that ERROR is a terminal state, so I do not see why somebody should wait for NOT_RUNNING when the client is in ERROR and close() returns true.

Alright I adjusted the close response to align with this.

mjsax · 2020-12-22T23:58:42Z

If the global thread dies, should we not transit to ERROR at the end?

It will. We don't strictly need to test it here as that is tested elsewhere but we can add it for clarity

I guess it would not hurt. Leave it up to you than.

mjsax · 2020-12-23T00:01:36Z

Why not just change this from CLOSE_CRASHED to CRASH ?

mjsax · 2020-12-23T00:02:50Z

Should we extend it to have two state transitions, RUNNING -> PENDING_ERROR and PENDING_ERROR -> ERROR ?

Actually it could probably be just PENDING_ERROR -> ERROR because it could be in RUNNING or REBALNCING previously

mjsax · 2020-12-23T00:09:59Z

mjsax · 2020-12-23T00:10:11Z

mjsax · 2020-12-23T00:15:41Z

Nit: the error message is weird -> Streams didn't transit to ERROR state.

mjsax

Made a pass over it.

wcarlson5

@mjsax I believe that I got to all your comments except for the shouldUpgradeFromEosAlphaToEosBeta tests one

wcarlson5 · 2021-01-04T16:38:16Z

yay ascii art! good catch I forgot about that

When I build the docs locally it looks about right

wcarlson5 · 2021-01-04T17:03:52Z

I don't think that we should return true. A user would then expect that that the state would then change to NOT_RUNNING and could be stuck waiting on that. Where if we return false in then they might retry and get stuck there.

I don't know which is a better problem to have but I think that changing the meaning of this return value won't add any more clarity to the situation.

wcarlson5 · 2021-01-04T17:27:21Z

It will. We don't strictly need to test it here as that is tested elsewhere but we can add it for clarity

wcarlson5 · 2021-01-04T17:30:06Z

It made it so that the new default was not used until we updated the Error transition as we are doing in this PR. :)

wcarlson5 · 2021-01-04T17:38:57Z

My thoughts where that we are calling close and so its is going to just be closed as the state transitions are cleared. This is the test that made me want you to review the PR I was worried that I might have broken the integrity of the tests as it is not behaving as I expected. However no other test was having this problem so I was not sure if I was understanding it correctly.

mjsax · 2021-01-07T00:02:19Z

I thought we always need to call close? If an error happens, we call the handler, and if the handler return shutdown, we transit to PENDING_ERROR. On close() we transit from PENDING_ERROR -> ERROR?

Or do I have some misconception?

the handler will call close, but the user will not need to. The PENDING_ERROR state is indicating the resources are closing before the transition to ERROR after which no more work will be done. We made it so the user can call close on PENDING_ERROR or ERROR but it will only log a warning

Thanks for clarifying!

wcarlson5

@mjsax I am still looking into the shouldUpgradeFromEosAlphaToEosBeta test. but not making a ton of progress I will try to carve out a little time tomorrow for it.

I think there was a bit of miss-understanding about what PENDING_ERROR means but I hope I cleared it up the comments below.

wcarlson5 · 2021-01-07T21:52:36Z

the handler will call close, but the user will not need to. The PENDING_ERROR state is indicating the resources are closing before the transition to ERROR after which no more work will be done. We made it so the user can call close on PENDING_ERROR or ERROR but it will only log a warning

wcarlson5 · 2021-01-07T21:56:20Z

the call to close() will have no effect. The PENDING_ERROR -> ERROR transition is like the PENDING_SHUTDOWN -> NOT_RUNNING where the transition from the PENDING... state is automatic

wcarlson5 · 2021-01-19T22:10:07Z

-            Arrays.asList(
-                KeyValue.pair(KafkaStreams.State.ERROR, KafkaStreams.State.PENDING_SHUTDOWN),
-                KeyValue.pair(KafkaStreams.State.PENDING_SHUTDOWN, KafkaStreams.State.NOT_RUNNING)
+                KeyValue.pair(State.PENDING_ERROR, State.ERROR)


Using just PENDING_ERROR -> ERROR because the transition to PENDING_ERROR can be from multiple sources. Also that transition is already tested so this check implies it

mjsax

Just some final nits. Overall LGTM.

mjsax · 2021-01-21T21:14:12Z

            () -> !observed.isEmpty() && observed.get(observed.size() - 1).value.equals(State.RUNNING),
            MAX_WAIT_TIME_MS,
-            () -> "Client did not startup on time. Observers transitions: " + observed
+            () -> "Client did not have the expected state transition on time. Observers transitions: " + observed


Why this change? We do wait for RUNNING?

We do wait for running, I was thinking of bringing it to match with the other methods below but that doesn't make it anymore useful so I will revet it.

mjsax · 2021-01-21T21:42:58Z

Btw: to what extend do we need to update the docs? We should at least add a section to streams/upgrade_guide.html to mention the change.

wcarlson5 · 2021-01-21T22:38:53Z

I extracted the https://github.com/apache/kafka/pull/9720/files#r562203226 fix to #9948

wcarlson5 · 2021-01-25T20:25:26Z

@mjsax it looks like there is one unrelated failure java.lang.RuntimeException: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'primary.test-topic-2' already exists.

glasser · 2023-04-11T19:22:00Z

Am I correct in believing this change was released in Kafka 2.8.0? I can't find it in the release notes. We just had a minor outage because we didn't realize when upgrading to v2.8 that some error handling/shutdown code in our Kafka Streams app couldn't assume that it only had to kick in when the state transitioned to NOT_RUNNING but also now has to handle ERROR too.

wcarlson5 commented Dec 10, 2020

View reviewed changes

mjsax added the streams label Dec 11, 2020

wcarlson5 changed the base branch from trunk to 2.7 December 11, 2020 21:30

wcarlson5 changed the base branch from 2.7 to trunk December 11, 2020 21:30

mjsax reviewed Dec 22, 2020

View reviewed changes

mjsax reviewed Dec 23, 2020

View reviewed changes

Comment thread streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java Outdated

mjsax reviewed Dec 23, 2020

View reviewed changes

Comment thread streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java Outdated

mjsax reviewed Dec 23, 2020

View reviewed changes

Comment thread ...ava/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java Outdated

mjsax reviewed Dec 23, 2020

View reviewed changes

wcarlson5 commented Jan 4, 2021

View reviewed changes

mjsax reviewed Jan 7, 2021

View reviewed changes

wcarlson5 commented Jan 7, 2021

View reviewed changes

mjsax added the kip Requires or implements a KIP label Jan 8, 2021

mjsax mentioned this pull request Jan 14, 2021

KAFKA-10017: fix 2 issues in EosBetaUpgradeIntegrationTest #9733

Merged

3 tasks

wcarlson5 added 9 commits January 19, 2021 11:15

init commit, one test is still failing

c7412f8

debug

42aa534

fix state store wipe

7c3e014

undo dep

ae8ea32

undo dep

9aff9a7

quick fixes

836e9ae

rebase replace thread so global threads close correctly

280dab1

init commit, one test is still failing

4e0b565

debug

5f73045

wcarlson5 added 7 commits January 19, 2021 13:51

fix state store wipe

e1b6bb0

undo dep

468a0a6

undo dep

9603659

quick fixes

0304647

address comments

a0a51d9

returned checks

35241f4

update fail msg

8078e96

wcarlson5 commented Jan 19, 2021

View reviewed changes

wcarlson5 added 2 commits January 19, 2021 14:11

update fail msg

de99a7c

close semantics

93beb2f

mjsax reviewed Jan 21, 2021

View reviewed changes

nits

0345202

extract system time change fix

e235b62

mjsax merged commit 647c609 into apache:trunk Jan 26, 2021

wcarlson5 deleted the wcarlson-fsm-changes branch January 26, 2021 19:54

Conversation

wcarlson5 commented Dec 9, 2020

Committer Checklist (excluded from commit message)

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Dec 22, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

mjsax Dec 22, 2020 •

edited

Loading

wcarlson5 left a comment •

edited

Loading