After KIP-360 this would not be thrown, but anyhow the caller like stream thread should always try to capture it since it cannot guarantee to always talk to a 2.5+ broker.

Just a reminder of the possible exception thrown.

Makes sense, but at the same time, it's a little confusing because InterruptException is a KafkaException. Maybe this is no big deal, though, if the intent of the doc is to say that you should treat an InterruptException specially.

Yes that's indeed the intention, to treat interrupt exception specially.

This is a minor change: since the exception maybe thrown BEFORE the task is created now I've decided to just encode the task-id, which means we cannot print the full task topology but I think it is not a huge step backwards -- actually I think it is cleaner on log4j to not print the full topology after a second thought.

+1

prop: +1 for the motivation, however could we make this a separate minor PR to reduce the current PR size?

@abbccdda My current proposal is to move all the PRs into a separate K9113-base branch which is periodically rebased on trunk, and have consecutive PRs against that base branch, so that we do not have to make sure each PR leaves the branch in a perfect state (system tests, unit tests, etc). So I think in this case this PR's size is okay -- as I said although it is big in LOC the non-testing changes are not so big, and this change is actually needed since I refactored the place when record collector etc are constructed.

Push the stateMgr.flush to individual functions of tasks.

This is another cleanup: clean and isZombie are now consolidated into one flag.

My thoughts around those function cleanup are: after we've added the state into Task, then we would not have such functions any more but just a single close function that depends on the state: e.g. if it is running we need to 1) close topology, 2) flush state / collector, 3) commit (if clean), and then 4) close state; if it is suspended we only need to do step 4).

SGTM. I agree it makes sense to tackle that in a separate PR.

This is another minor change: now we always delete the checkpoint file after initialization and not rely on the eos flag (since it is removed); for non-eos it is not a huge regression since the first commit after initialization would write one checkpoint so the only "vulnerable window" is during the initialization and the first commit, which is around 100ms --- but maybe I will add this back in further cleanup after I've straight-out the state manager.

I'm fine with removing this for now and planning to consider how best to re-introduce it later. Not sure if I totally buy the idea to have some period of time where the file doesn't exist, but I'm happy to re-consider it later.
Can you create a follow-up task, so we don't forget about it?

Yes -- actually my plan is to not change this behavior inside the scope of KAFKA-9113, but this PR took a step backwards (I kept all of those changes in mind) to clean up some messy dependencies, but in a later PR we could add this logic back after we've figured out the whole hierarchy.

prop: +1 for the motivation, however could we make this a separate minor PR to reduce the current PR size?

q: what's the benefit of checking the config flag every time versus one time initialization?

Ack.

I tried to abstract the EOS logic away from task, but due to an optimization it ended up still needed to be kept inside the class, so I've decided to add it back.

req: this member should be removed.

prop: this comment is not necessary. If we want to enforce the transactionInFlight correctness, we should add a validation for all transaction related private functions as

if (!eosEnabled && transactionInFlight) {
  throw new IllegalStateException();
}

We indeed check eosEnabled everywhere we condition on transationInFlight.

req: I don't think we should call maybeBeginTxn, as we do that call during every send. If we are not in a transaction but calling commit(), that sounds like an illegal state to me, or we should just bypass the whole commit logic as it indicates we didn't do any send call in the past when EOS is turned on.

I thought about that but I think this is still a legal state, consider this: you processed some records but no outputs sent via the changelogger (i.e. send never called), then you call commit. In this case this transaction contains no data on Kafka but the local states may still be updated and the consumed offsets still advanced, in this case you still need to call 1) beginTxn, 2) sendOffsetsToTxn and then 3) commitTxn. Right?

prop: suspend and closeSuspended could cause confusions as the meaning was very close. Maybe name closeSuspended to closeSuspendedIntervalState

Yes, my plan is to eventually remove closeSuspended at just let close to act differently based on state.

req: typo unknown Pid

Ack

Why we use a super class of UnknownProducerIdException here?

Just to not lose generality: UnknownProducerIdException is actually the only extended class of OutOfOrderSequenceException, and other scenarios that the more general OutOfOrderSequenceException is thrown directly is for starting / end offset falling out of range. So I think letting streams to handle the general OutOfOrderSequenceException is fine too.

prop: for the exception tests, could we build a template test struct like

testSuite {
   mockProducer;
  expectedExceptionType;
}

to reduce the amount of test code?

Just my 2 cents: having a lot of factored-out code in tests usually hinders, rather than helps, maintainability. In the long run, the overall number of lines in the test file doesn't hurt anything, because you rarely sit down to read all the methods (typically, just while doing the review like this). After this PR is merged, you would almost always just be trying to read and understand a single method.

Thus, it pays to optimize for single-method legibility. Having a test harness to go read, and other support methods to go read, just to understand this method is only going to get in the way. As it is right now, this method is 28 lines long, perfectly legible and clear. Trading clarity for de-duplication is a bad deal.

Prop: we need to add some comments for removed large tests, for example in high level what are they testing, what functionality has been migrated and how it's ensured the same coverage to make review easier.

+1 I personally decided not to worry about it, but just keep a mental todo to come back after all the cleanups are complete to revisit test coverage (lines and semantics). But it would be nice to know why it's ok that they were removed.

@abbccdda yeah my plan is to keep each PR against a separate base branch so that we do not need require each PR to be a "transaction" that always leave the branch still in a consistent and perfect state.

I will add TODO to come back after all the cleanups for revamping our unit test coverage.

Makes sense, but at the same time, it's a little confusing because InterruptException is a KafkaException. Maybe this is no big deal, though, if the intent of the doc is to say that you should treat an InterruptException specially.

+1

SGTM. I agree it makes sense to tackle that in a separate PR.

It's not clear from context what this does, and hence why it's necessary here. Can you add an explanatory comment (or change clear to something more... clear)?

Ack.

I initially had some confusion here about whether suspend could throw in an unclean close, leading to closeSuspended never getting called. Then I went back and read the comment... Maybe we can simplify all these code paths if unclean invocations never throw at all, at the lowest levels. Then, we could get rid of all these catch blocks up the hierarchy. WDYT?

Yes, sounds good to me!

One tricky thing is that in commit / suspend there are several steps to execute, and when clean = false we should still execute all the steps even if early steps throw.

Didn't follow this last thought. If we have a pattern where methods with clean=false never throw (which was my suggestion), it would automatically do what you are saying here, right?

There are two things with clean=false here: 1) we should not throw any more, and 2) we can skip some steps like trying to commit, because we know it is likely to fail, and even it does not it's not meaningful anyways. 2) is an optional optimization admittedly.

If it's no longer used, we should remove it.

Just my 2 cents: having a lot of factored-out code in tests usually hinders, rather than helps, maintainability. In the long run, the overall number of lines in the test file doesn't hurt anything, because you rarely sit down to read all the methods (typically, just while doing the review like this). After this PR is merged, you would almost always just be trying to read and understand a single method.

Thus, it pays to optimize for single-method legibility. Having a test harness to go read, and other support methods to go read, just to understand this method is only going to get in the way. As it is right now, this method is 28 lines long, perfectly legible and clear. Trading clarity for de-duplication is a bad deal.

+1 I personally decided not to worry about it, but just keep a mental todo to come back after all the cleanups are complete to revisit test coverage (lines and semantics). But it would be nice to know why it's ok that they were removed.