KAFKA-9441: Add internal StreamsProducer

[mjsax]

Upfront refactoring for KIP-447.

Introduces StreamsProducer that allows to share a producer over multiple tasks and track the TX status.

Call for review @guozhangwang @abbccdda

[mjsax]

This allow us to simplify RecordCollectorImplTest significantly -- we can register a exception that we want to throw on the next invocation of the corresponding method.

[mjsax]

I think we should do the regular state check first before we throw any test ingested exception.

[abbccdda]

That depends on the purpose of this class. According to the meta comment:

By default this mock will synchronously complete each send call successfully. However it can be configured to allow the user to control the completion of the call and supply an optional error for the producer to throw.

We are definitely not concerned too much about the internal correctness of MockProducer, as long as it is reproducing the error we injected IMHO.

[mjsax]

We just replaced this with the commitTransactionException -- a user can just set a ProducerFencedException to get the same behavior

[mjsax]

We use StreamsProducer now that encapsulates the actually producer, does transaction state handling, and error handling. RecordCollectorImpl is now agnostic to transactions as desired.

[mjsax]

Note that we don't even need to call a "maybeInitTx" on StreamsProducer -- it embeds the corresponding logic internally and fully automatic.

[mjsax]

This comment is unclear to me, that is why I added the text -- can you explain what it means @guozhangwang ?

[guozhangwang]

These first line should be removed -- I did it in my current on-going PR.

[mjsax]

Just some side cleanup :)

[abbccdda]

This is nice!

[mjsax]

I think we could also set to null here -- not sure which one would be better?

[mjsax]

Seems the comment does not make sense any longer? Only side cleanup in this class

[mjsax]

To avoid redundant code, we setup everything for non-eos and eos case upfront (even if not every test needs everything).

[mjsax]

We can avoid to create "collectors" in almost all tests now reducing the overall complexity of the test

[mjsax]

I added stricter verification to all tests, to make sure the right exception was triggered.

[mjsax]

MockProducer does not support the correct handling of callback exceptions, hence, we need to do it manually. (Did not think it worth to extend MockProducer -- let me know what you think)

[mjsax]

The name and the test does not really match as it throws in beginTransaction. In any case, we have test for both cases already and thus I removed this duplicate test (please verify that I did not reduce test coverage)

[guozhangwang]

The two cases are differ that one throwing KafkaException (fatal) and the other throwing ProducerFencedException (task-migrated).

[mjsax]

My argument was that the old test was named "send" but did throw in "beginTx" (and there was already a corresponding "beginTx" test in the old code -> hence "redundant" / or at least need a fix). However, the new StreamsProducerTest should cover all cases for "send" as well as "beginTx":

BeginTx:

• shouldThrowTaskMigrateExceptionOnEosBeginTxnFenced
• shouldThrowTaskMigrateExceptionOnEosBeginTxnError
• shouldFailOnEosBeginTxnFatal

send:

• shouldThrowTaskMigratedExceptionOnEosSendFenced
• shouldThrowTaskMigratedExceptionOnEosSendUnknownPid
• shouldThrowTaskMigrateExceptionOnEosSendOffsetFenced
• shouldThrowStreamsExceptionOnEosSendOffsetError
• shouldFailOnEosSendOffsetFatal

[mjsax]

This is actually an incorrect test -- on commit, we would never do anything with StandbyTasks because they always return false on commitNeeded() -- the used StateMachineTask got a corresponding fix.

[guozhangwang]

Hmm... maybe that's correct, but that would also mean that the standby tasks are not committed at any occasions except closing which sounds like a lurking bug to me?

[mjsax]

Sounds correct -- I was double checking this code and this issue was introduced in the refactoring via 4090f9a

To not block this PR further, I would suggest to keep this change as-is and do a follow up PR to fix the issue.

[guozhangwang]

I will make a separate PR to fix this.

[mjsax]

Removed this test for the same reason as above.

[mjsax]

This is the fix for StateMachineTask -- it should behave the same way as an actual active or standby task

[mjsax]

After John's TDD fix, this is not used any longer (just cleanup to remove it).

[guozhangwang]

Are the test failures in org.apache.kafka.streams.TopologyTestDriverTest relevant?

[guozhangwang]

Thanks @mjsax for this PR! I made a pass on the non-testing part and it generally looks promising to me. As for merging it I'd suggest we do it after #8058 since it resolved several exception handling TODOs as in RecordCollector so it may make this PR smaller.

[guozhangwang]

nit: can we consolidate producerFencedOnCommitTxn to the more-general commitTransactionException? I.e. if you want to fence on commit, you just register the commitTransactionException as a ProducerFencedException

[guozhangwang]

The passed in value would be true for both alpha and beta right? if yes we can just name it eosEnabled.

[guozhangwang]

These first line should be removed -- I did it in my current on-going PR.

[guozhangwang]

nit: taskProducers

[guozhangwang]

When are we removing the entry upon task closure? If it never cleans up we could potentially have an ever-growing map.

[mjsax]

The TaskManager does this in handleAssignment() and handleLostAll().

[guozhangwang]

Regret to see this ... I guess in order to allow mocking the task creation inside threadTest we'd have to do this, but it also makes the code less readable. Maybe we can augment the createTask interface to pass in all parameters so that the preparation logic can then be pushed into the StreamThread?

[guozhangwang]

Also the taskProducers map is reference in three different classes: 1) creator to populate it, 2) task-manager to remove from it, 3) thread to get metrics from it (this is to be fixed).

I think we can let it to be purely owned by the task-manager, after we fixed 3) (cc @vvcephei ), as for 1) if we can augment the createTask above, we can then push the addition logic into TaskManager and creates the record-collector with the passed-in producer, and the only pass-in the record collector into createTask interface.

WDYT?

[mjsax]

I did not want to do more refactoring this PR is already quite big -- it would be best to push TaskCreator stuff completely into TaskManager.

[guozhangwang]

Fair enough :) could you add a TODO marker here?

[guozhangwang]

nit: in KIP-447 we would not only use applicationId but also consumer's whole ConsumerGroupMetadata, so for future-uses I'm thinking maybe we can just pass in the consumer object here, and then before KIP-447 we can just get the applicationId from its consumer.metadata.

[mjsax]

It's for sure an intermediate state -- will get changed in follow up PRs.

[guozhangwang]

The passed in task-id are used for two purposes: 1) define eosAlphaEnabled, 2) use in logging (btw there are still some place where we do not check taskId == null). And in #8058 we no longer log the task-id in TaskMigrated since when it happens we would migrate all the tasks anyways. So we do not need it in the TaskMigrated exception.

So I'd suggest the following:

a) we keep a String for taskIds which would be a single task-id if the passed in taskId != null or default to "all owned active tasks" for logging purposes only.
b) we keep the eosEnabled checking the taskId as is.

And moving forward as we remove the alpha we would replace this parameter with a boolean only since it would always be all owned active tasks.

[guozhangwang]

Ditto here: even later when we introduce the eosBeta the logic here would be the same as we wrap the logic in TransactionProducer right? In that case we can just name this boolean as eosEnabled.

[guozhangwang]

nit: rename to isProducerFenced?

[mjsax]

An UnknownProducerIdException does not imply fencing? Does it?

[guozhangwang]

Well yes and no: it is not fenced by txn coordinator, but by broker :) Anyways I'm just a feeling a bit awkward that in the caller we say if it is recoverable, we always handle it as task-migrated --- it is a very nit so your call

[mjsax]

It was called isRecoverable() before -- think I just leave it.

[mjsax]

Updates this PR -- also fixed the TDD issue.

[mjsax]

Rebase to resolve merge conflict. Call for review @guozhangwang

[abbccdda]

Take an initial pass. I put an architectural idea there which we could discuss further after the long weekend.

[abbccdda]

This is nice!

[abbccdda]

nit: could be simplified as eosEnabled

[abbccdda]

That depends on the purpose of this class. According to the meta comment:

By default this mock will synchronously complete each send call successfully. However it can be configured to allow the user to control the completion of the call and supply an optional error for the producer to throw.

We are definitely not concerned too much about the internal correctness of MockProducer, as long as it is reproducing the error we injected IMHO.

[abbccdda]

This is not used.

[abbccdda]

"Task " + taskId + " could not get partition information for topic " + topic

[abbccdda]

It looks weird to let transactionManager to send record when the application is not on EOS. What's the issue with name StreamProducer?

[mjsax]

We can also go back to StreamProducer -- I just had the impression that TransactionManager is the better name. You are correct, that if EOS is disabled, the TM has nothing to manage :) -- I don't feel strong about the name either way.

[abbccdda]

should be abortTransactionIfOngoing or maybeAbortTransaction as there may not be one on-going txn.

[mjsax]

Well, the idea of the TM is "hide" those details -- as a caller, I don't want to think about it, I just say, something was wrong, please abort. That the call might be a no-op should be an implementation detail to the caller.

[abbccdda]

I don't think the comment is necessary

[mjsax]

Agree -- copied this from RecrodCollectorImpl...

[abbccdda]

On stream this could only be consumer, so we could use consumerGroupId

[mjsax]

appid == groupid and in KS we always use appid.

[abbccdda]

nit: as we are using full names elsewhere, we could also name it maybeInitTransaction

[mjsax]

Thanks for the review @abbccdda -- updated this PR.

[mjsax]

Needed to change the constructor to make it shorter (checkstyle failed). Hence, some variables cannot be final any longer.

[mjsax]

Java 8: org.apache.kafka.streams.integration.RegexSourceIntegrationTest.testRegexMatchesTopicsAWhenDeleted

Java 11: org.apache.kafka.streams.state.internals.RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest.shouldForwardAllDbOptionsCalls

(No need to retest atm. Waiting for reviews.)

[guozhangwang]

retest this please

[abbccdda]

I couldn't find one of my important previous comments anywhere, so I will just rephrase it here:
I could see that we still need to keep eosEnabled flag inside RecordCollector, and we are overusing TransactionManager for a lot of non-transaction related work such as send, flush, etc.

I'm thinking whether it makes sense to define a ProgressCommitManager instead to take in both Producer and Consumer, and the only time it shows up in RecordCollector is to do the commit. This way we could internalize all the EOS related logic inside ProgressCommitManager and record collector shall still own normal producer operations. And collector should no longer be aware of EOS logic.

WDYT?

[guozhangwang]

@mjsax I made another pass (this time including the tests) and most are minor. Please feel free to merge after you've replied them.

[guozhangwang]

We should remove K9113 here since in the scope of KAFKA-9113 we are going to treat this as fatal (which is already incorporated now). If we want to use the buffer-and-retry logic then it would be out of 9113's scope.

[guozhangwang]

I think each transaction manager only manages one producer right?

Also nit: I feel more inclined to the name of StreamProducer since it is used for both eos and non-eos, @abbccdda @mjsax wdyt?

[guozhangwang]

Well yes and no: it is not fenced by txn coordinator, but by broker :) Anyways I'm just a feeling a bit awkward that in the caller we say if it is recoverable, we always handle it as task-migrated --- it is a very nit so your call

[guozhangwang]

nit: If we rename it to StreamProducer then maybe this can be renamed to embeddedProducer or sth.

[mjsax]

Are you fine with kafkaProducer() ?

[guozhangwang]

Yup.

[guozhangwang]

If the serialization failed we wrap it as a StreamsException -- with this PR it seems a regression that we do not do it any more.

[guozhangwang]

Brackets looks unnecessary here.

[mjsax]

I reuse the variable name thrown twice and want to make it final -- hence the braces

[guozhangwang]

qq: not sure what does this mean.

[mjsax]

It means functional test, ie, we that something should work (instead of negative test that test that something should not work/throw an error). Are you fine with this or do you have a suggestion on how to rephrase it?

[guozhangwang]

Yeah this is fine.

[guozhangwang]

nit: shouldFailOnInitTxnWhenEOSDisabled.

[guozhangwang]

Ditto here. Not sure what does positive mean.

[guozhangwang]

nit: eosEnabled.

[abbccdda]

Blocking the merge here as we want to at least rename the class to StreamProducer.

[mjsax]

Update this PR. Let me know if I addresses all comments sufficiently. If yes, I hope we can merge if Jenkins is green.

[guozhangwang]

LGTM.

[mjsax]

Rebase this PR to resolve merge conflicts.