title: Getting java.lang.ClassCastException: [B cannot be cast to org.springframework.messaging.Message exception after consuming batch
body: <p>I am using spring-cloud-stream-kafka-binder-3.0.4 to consume the messages in batch, after consuming, converting the Message into object but getting the above exception.</p>
  <p>Here is the code:</p>
  <pre><code>@StreamListener(ActivityChannel.ACTIVITY_INPUT_CHANNEL)
  public void handleActivity(List&lt;Message&lt;Event&gt;&gt; messages,
      @Header(name = &quot;deliveryAttempt&quot;, defaultValue = &quot;1&quot;) int deliveryAttempt,
      @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment)  
  {
      try
      {
          log.info(&quot;Received activity message with message length {} attempt {}&quot;,
                messages.size(), deliveryAttempt);
          List&lt;Event&gt; eventList = messages.stream().map(Message::getPayload).collect(Collectors.toList());
          nodeConfigActivityBatchProcessor.processNodeConfigActivity(eventList);
          acknowledgment.acknowledge();
          log.debug(&quot;Processed activity message {} successfully!!&quot;, messages.size());
      } 
      catch (Exception e)
      { 
          throw e;
      }
  }
  </code></pre>
  <p>Configuration:</p>
  <pre><code>spring.cloud.stream.bindings.activity-input-channel.destination=TOPIC.FEED.NAME
  spring.cloud.stream.bindings.activity-input-channel.contentType=application/json
  spring.cloud.stream.bindings.activity-input-channel.consumer.batch-mode=true
  spring.cloud.stream.bindings.activity-input-channel.consumer.max-attempts=1
  spring.cloud.stream.kafka.bindings.activity-input-channel.consumer.auto-commit-offset=false
  spring.cloud.stream.kafka.bindings.activity-input-channel.consumer.reset-offsets=true
  spring.cloud.stream.kafka.bindings.activity-input-channel.consumer.start-offset=latest
  spring.kafka.consumer.max-poll-records=5
  spring.kafka.consumer.fetch-max-wait=60000
  spring.kafka.consumer.fetch-min-size=500
  </code></pre>
  <p>I get the above error at this line <code>List&lt;Event&gt; eventList = messages.stream().map(Message::getPayload).collect(Collectors.toList());</code> at .collect(Collectors.toList()). I am not able to figure out why??</p>
  <p>if I check <code>Message&lt;Event&gt; eventMessage = messages.get(0)</code> getting the same exception(messages is the list of Message variable).</p>
  <p>if batch mode if false then it only consume a single message handleActivity(Message message), then it works fine, no exception.</p>
  <p>Is there any deserializer needs to be added when using batch mode???</p>