lct45 commented on pull request #10813: URL: https://github.com/apache/kafka/pull/10813#issuecomment-858849165
Okay I did some digging -> `defaultKeySerde` and `defaultValueSerde` are only called from the `init` of `AbstractProcessorContext`. I checked all the places that we call `AbstractProcessorContext#keySerde()` and `AbstractProcessorContext#valueSerde()` to make sure we're catching all the potential NPEs and I am fairly confident that we're ok. I did some streamlining so now we throw the `ConfigException` right after we access `AbstractProcessorContext#keySerde()` / `valueSerde()` so we aren't passing null's around and there's some tracking b/w throwing errors and calling a certain method. The one place this wasn't possible, was with creating state stores. Right now, we pass around `context.KeySerde()` and `context.valueSerde()` rather than just the `context` in `MeteredKeyValueStore`, `MeteredSessionStore`, and `MeteredWindowStore`. The tricky part with moving to passing around context is that we need to accept two types of context, a `ProcessorContext` and a `StateStoreContext`. I'm open to either leaving these calls as less streamlined than everything else, or duplicating code in `WrappingNullableUtils` to accept both types of context. Thoughts @mjsax @ableegoldman ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
