For Kafka Streams, if we set the commit.internal.ms config to Long.MAX_VALUE this will effectively avoid that Kafka Streams commits automatically but commit only after calling lower level processor api context.commit(), we can control to commit or not? If yes then in case of problem or message contain malformed format how we can remove those message from topic?
Resources: https://docs.confluent.io/current/streams/index.html
You can not remove a message from a topic... Kafka topics are append-only immutable logs.
Also note, if you commit a message, it is not remove from the topic either. Kafka is not a message queue --- committing offsets only implies to track the current progress (ie, offset) in the log, but it does not delete any messages. (You can think of it just as a "cursor".)
Thus, if a message is malformed and you don't want to process it, you can just skip over it. There is actually no reason to set a large commit interval for this case and do manual commits.