KAFKA-13654: Extend KStream process with new Processor API

[jeqo]

KIP-820 implementation.

Changes:

• Implement changes proposed in the KIP.
• Adapt existing methods to implement FixedKeyProcessor instead of Processor, to guarantee fixed keys.

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)

[vvcephei]

Thanks, @jeqo ! Great work, as always. I've attached some comments, but not too many. I think we should be able to merge soon.

I didn't double-check that all the APIs in this PR are exactly as represented in the KIP. I'll be sure to do that before merging.

[vvcephei]

Hmm, why is this one named NamedInternal.empty(), but the other one (on L1515) named Named.as(builder.newProcessorName(PROCESSOR_NAME))?

We should be really sure that we get the right processor naming behavior. It's extremely difficult to change them later, so if we get it wrong, we basically have to support the wrong behavior forever with a bunch of special-case logic.

[jeqo]

Great catch! Do you think a new processor name is needed for processValues? e.g. KSTREAM-PROCESSVALUES-

[vvcephei]

Yeah, that's probably a good idea, actually.

[jeqo]

New prefix added 👍🏽

[vvcephei]

Are we missing adding the oldProcessor to the topology here?

[jeqo]

Interestingly enough, it seems that the old processor is not being added to the topology builder anymore.
Internal Topology doesn't even have a #addProcessor(String, OldProcessorSupplier, ...) method anymore.
Do you think there might be some regression here?

[vvcephei]

Hmm. I seem to recall having refactored the internal topology builder code a long time ago so that the old processors would get shimmed to the new API so that we only need one path.

I guess the task here would be to double-check that adding old processors still works without attached stores, with attached stores via the addProcessor method, and with attached stores via the old ProcessorSupplier.stores().

In addition to that, I'd also double check that this code block is actually used by the last of those code paths.

Of both of those check out, then I think we have a correct implementation here.

[jeqo]

Added a couple of tests to KStream to test these.

[vvcephei]

Why not just call delegate.process(record)? Actually, why do we need to add this method here at all?

[vvcephei]

Oh, now I see. Why not just capture the result of ProcessorAdapter.adapt(mockProcessor) in the constructor and use that for the this.mockProcessor field instead? That way it would already be the right type. That's basically what we were doing before, although I agree we don't need to have that method to expose the processor on the super class.

[jeqo]

Got it. Replaced the initialization with: this.mockProcessor = (MockApiProcessor<...>) ProcessorAdapter.adapt(mockProcessor);
Let me now if this look good now.

[jeqo]

Fallback to MockProcessor instead of MockApiProcessor as there is no need to migrate this I think.

[vvcephei]

Oy, forgot about the scala API ... again. Do we need to add the new interfaces to the scala API? Also, the scala deprecated annotation takes parameters about when it was deprecated and what you should use instead, which we should make use of.

[jeqo]

Got it. Params are added. I also forgot to add processValues to the Scala API. Should be there now.

[vvcephei]

This wasn't in the KIP... Do we need it for something?

[jeqo]

These are required to implement from ProcessingContext#currentSystemTimeMs() and ProcessingContext#currentStreamTimeMs(). I followed a similar approach as

kafka/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java

Lines 401 to 407 in 0d518aa

	public void setCurrentSystemTimeMs(final long currentSystemTimeMs) {
	this.currentSystemTimeMs = currentSystemTimeMs;
	}
	public void setCurrentStreamTimeMs(final long currentStreamTimeMs) {
	this.currentStreamTimeMs = currentStreamTimeMs;
	}

.

[vvcephei]

Ok, I think I remember this now. The KIP that added those time APIs forgot to add them to the new Processor API and only did the old API. I think it got missed because both KIPs were going in parallel.

Can you dig up that KIP and just notify the vote thread of it that we need to amend the KIP to address this oversight? It shouldn't require a new vote or anything, but we should update the KIP.

[jeqo]

I see. I have updated the KIP and the vote thread 👍🏽

[jeqo]

Thank you, @vvcephei !! I have applied most suggestions, let me know your thoughts 🙌🏽

[vvcephei]

Looks good to me! Thanks, @jeqo .

It looks like there's a checkstyle failure, and I had a small comment.

[vvcephei]

Just a note: Unlike in the Java API, where you could not have assigned the void result of the method to a variable, in the Scala API, someone could actually have assigned the result of this method to a Unit-typed variable. They still wouldn't have been able to do anything with it, so I think the chances are small they actually would have captured the result, and I think that in this case, it's better to make this small change, rather than add two new methods (which would be different from the Java counterparts) for 100% compatibility.

[vvcephei]

Gracias!