KAFKA-6775: Fix the issue of without init super class's

[jiminhsieh]

Some anonymous classes of AbstractProcessor didn't initialize their superclass. This will not set up ProcessorContext context at AbstractProcessor.

Committer Checklist (excluded from commit message)

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

[mjsax]

Thanks for the PR.

I just double checked the code, and it seems that AbstractProcessor is use on other places, too. Did you double check those, if super.init(context) is called there?

It might also make sense to make AbstractProcessor#context protected?

[mjsax]

might be better to remove the overwrite method completely?

[hachikuji]

Alternatively, if we don't care about the context, we could just use Processor directly instead of AbstractProcessor.

[jiminhsieh]

Thanks for the suggestions! After saw this snippet:

kafka/streams/test-utils/src/test/java/org/apache/kafka/streams/MockProcessorContextTest.java

Lines 45 to 50 in 886daf5

	final AbstractProcessor<String, Long> processor = new AbstractProcessor<String, Long>() {
	@Override
	public void process(final String key, final Long value) {
	context().forward(key + value, key.length() + value);
	}
	};

I wonder how about only keeping process this overwrite method and removing the other overwrite methods which include init, punctuate, and close? I think it is much clear.

[hachikuji]

Sounds good to me.

[mjsax]

as above

[mjsax]

nit: better to do this as first instruction?

[mjsax]

\cc @bbejeck @vvcephei

[jiminhsieh]

Thanks for the reviewing and suggestion!

I just double checked the code, and it seems that AbstractProcessor is use on other places, too. Did you double check those, if super.init(context) is called there?

If this meant other anonymous classes of AbstractProcessor like some of the usages at MockProcessorContextTest, those are simply didn't overwrite.

it might also make sense to make AbstractProcessor#context protected?

I am not sure about this since there is AbstractProcessor#context() which returns AbstractProcessor#context. It does the same job as making AbstractProcessor#context protected. It means we have two ways to access AbstractProcessor#context.

[guozhangwang]

LGTM!