[jira] [Comment Edited] (KAFKA-19923) Kafka Streams throws ClassCastException with different Consumed instances.

["sanghyeok An (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-19923?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18041039#comment-18041039 ] 

sanghyeok An edited comment on KAFKA-19923 at 11/27/25 11:38 AM:
-----------------------------------------------------------------

Could you let me know where would be a good place to document this guideline? I’d be happy to take a first pass at it.

Regarding your comment;

And yes, we could maybe add some code when merging nodes, to verify that we don't get conflicting configs passed in, and throw a better exception with a clear error message.

 

As we discussed earlier, there is currently no straightforward way to detect semantic conflicts among {{TimestampExtractor}} or Serdes implementations.

If Kafka Streams has any plan to support semantic equivalence checks for {{TimestampExtractor}} or Serdes in the future, please let me know. I would be glad to contribute.
However, I understand that such a change would likely be breaking. (Maybe 5.0.0???? )


was (Author: JIRAUSER303328):
*Could you let me know where would be a good place to document this guideline? I’d be happy to take a first pass at it.*

Regarding your comment;

And yes, we could maybe add some code when merging nodes, to verify that we don't get conflicting configs passed in, and throw a better exception with a clear error message.

 

As we discussed earlier, there is currently no straightforward way to detect semantic conflicts among {{TimestampExtractor}} or Serdes implementations.

If Kafka Streams has any plan to support semantic equivalence checks for {{TimestampExtractor}} or Serdes in the future, please let me know. I would be glad to contribute.
However, I understand that such a change would likely be breaking. (Maybe 5.0.0???? )

> Kafka Streams throws ClassCastException with different Consumed instances.
> --------------------------------------------------------------------------
>
>                 Key: KAFKA-19923
>                 URL: https://issues.apache.org/jira/browse/KAFKA-19923
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>            Reporter: sanghyeok An
>            Assignee: sanghyeok An
>            Priority: Minor
>
> Kafka Streams throws a ClassCastException when using different Consumed instances for the same topic.
> For example:
> {code:java}
> builder.stream("A", Consumed.with(Serdes.String(), Serdes.String()))
>        .peek((k, v) -> System.out.println(k));
> builder.stream("A", Consumed.with(Serdes.ByteArray(), Serdes.ByteArray()))
>        .peek((k, v) -> System.out.println(k));
> {code}
>  
> Since both use the same topic name and the same ConsumedInternal configuration for auto offset reset, these two StreamSourceNodes are merged during topology building.
>  
> As a result, the Topology is built successfully.
>  
> However, when the StreamThread starts, the consumer begins to receive records from the broker, and the records flow through the pipeline, a ClassCastException is thrown at runtime.
>  
> In my opinion, we have two options:
>  # Document this behavior.
>  # When merging source nodes, the builder should consider the full ConsumedInternal configuration (for example, key/value SerDes and timestamp extractor), instead of only the auto offset reset policy.
>  
> I think option 1 is also acceptable, because Kafka Streams will fail fast with a ClassCastException before the consumer commits any offsets.
>  
> Option 2 would require more substantial changes in Kafka Streams, because TimestampExtractor and key/value SerDes do not expose a straightforward way to check semantic equivalence.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)