create a map using key value of ConsumerRecord kafka

I am using KafkaConsumer to read messages from a list of kafka topics. I am using a custom Deserializer. below is the method from Deserializer:

public Object deserialize(String s, byte[] bytes) {
    if (!isEmpty(bytes)) {
      try {
        return objectMapper.readValue(bytes, new TypeReference<Map<String, Object>>() {
        });
      } catch (Exception e) {
        LOGGER.error("Error while deserializing bytes[] {} to json object ", new String(bytes), e);
      }
    }

    return null;
  }

No error happens here. Following is the screenshot of ConsumerRecords that I get:

The ConsumerRecord value tag has the values as a Map.
So following is my requirement:

• Use the messages from the topic to create a csv file.
• because there are multiple topics, so it's required to create multiple files using messages from each topic.

Probable Solution:

• separate the messages by topic. each topic can have a list of messages.
• so, I need a Map<String, List<Map<String, Value>>>. here the key(String) should be the topics and the value is :
  Value tag from ConsumerRecord in screenshot.

Questions:

1. How can I get the value tag from ConsumerRecord return as Map. Till now this is the best I could do:

   List<Object> customerMigrationKafkaMessages =
   consumerRecords.stream()
   .flatMap(c -> StreamSupport.stream(c.spliterator(), false).map(ConsumerRecord::value))
   .collect(Collectors.toList());

2. But, need to iterate ConsumerRecords in a way that I can create a Map. Key is the topic name, and value is the List object of the Map from above step.

3. Why do I need step1. as HashMap? 1. The keyset would serve as the header of the csv that I have to create. 2. The valueSet will the rows in the csv file.

Please help me with some ideas, I am not able to code this.
Thanks!

Edit 1:
Adding my comments to the answer from @OneCricketer

1. https://stackoverflow.com/questions/76437635/create-a-map-using-key-value-of-consumerrecord-kafka#:~:text=already%20has%20a-,JsonDeserializer,-class%20that%20returns - I have tried using this:

   kafka.value.deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
   But keep getting exception: Caused by: java.lang.IllegalStateException: No type information in headers and no default type provided

2. https://stackoverflow.com/questions/76437635/create-a-map-using-key-value-of-consumerrecord-kafka/76440122#76440122:~:text=structure%20of%20your-,JSON,-in%20order%20to : Yes, I know the structure. But, as I mentioned, there are multiple topics, so I will end up creating multiple POJO(which I did plan earlier), but later I am again converting those POJO to string and then writing to the file. Also, will have to create a Map before deserialization happens, so that I can use the right class type corresponding to the topic. Example:

Map myMap = new HashMap<>();
    myMap.put("TopicA", MyClassA.class);
    myMap.put("TopicB", MyClassB.class);
    public Object deserialize(String topic, String s, byte[] bytes) {
      if (!isEmpty(bytes)) {
        try {
          return objectMapper.readValue(bytes, myMap.get(topic));
        } catch (Exception e) {
          LOGGER.error("Error while deserializing bytes[] {} to json object ", new String(bytes), e);
        }
      }
      return null;

3. https://stackoverflow.com/questions/76437635/create-a-map-using-key-value-of-consumerrecord-kafka/76440122#76440122:~:text=YourCsvClass.class)-,Open,-a%20file%20handle - I might have misunderstood this.I can have a list of messages for each topic, and all those messages should go to a one file, and not file per message.
   So, if I have 4 topics processed. I need to have 4 files created.

> I am using a custom Deserializer

Kafka already has a JsonDeserializer class that returns JsonNode, and can be iterated the same as a Map.

> create a csv file

I'd recommend you use json lines.

But otherwise, you need to consider the fact that multiple consumer threads, or separate servers are running, so you'd end up with many files (unless you guarantee you have a single partition topic).

In any case, I assume you already know the structure of your JSON in order to actually write CSV headers?

So,

1. Create a POJO
2. Add Jackson csv dataformat dependency
3. Rewrite your Deserializer to return a POJO, rather than a Map (replace TypeReference with YourCsvClass.class)
4. Open a file handle before your consumer starts, or keep a Map<String, File> to store an open file per key, which you must flush each file handle when consumer batch is complete, and close each file handle on JVM shutdown (or after consumer batch)
5. Append message.value() to that file within the consumer loop. You do not need Streams here, nor does the Consumer iterator need to become a Map

I'll note that Kafka isn't really related to your problem, other than you shouldn't try to create one file per record. Refer https://www.baeldung.com/java-converting-json-to-csv

Also worth pointing out the Kafka Connect HDFS/S3 Sink can already do this for you, and will properly handle multiple partition topics (though the filename cannot be changed to use the record key)