Fix unstable batch test in MessageDrivenAdapterTests

By starting the adapter after sending messages, ensure a single poll retrieves the full batch messages.

Signed-off-by: Jiandong Ma <jiandong.ma.cn@gmail.com>

**Auto-cherry-pick to `7.0.x` & `6.5.x`**

Author: mjd507

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java

@@ -110,6 +110,7 @@
  * @author Cameron Mayfield
  * @author Urs Keller
  * @author Jooyoung Pyoung
+ * @author Jiandong Ma
  *
  * @since 5.4
  *
@@ -432,8 +433,6 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment a
 			}
 
 		});
-		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 1);
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		senderProps.put(ProducerConfig.LINGER_MS_CONFIG, 0);
@@ -443,6 +442,9 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment a
 		template.sendDefault(0, 1487694048607L, 1, "foo");
 		template.sendDefault(0, 1487694048608L, 1, "bar");
 
+		adapter.start();
+		ContainerTestUtils.waitForAssignment(container, 1);
+
 		Message<?> received = out.receive(10000);
 		assertThat(received).isNotNull();
 		Object payload = received.getPayload();
@@ -461,6 +463,11 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment a
 
 		assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
+		adapter.stop();
+
+		final CountDownLatch restartAssignmentLatch = new CountDownLatch(1);
+		adapter.setOnPartitionsAssignedSeekCallback((map, consumer) -> restartAssignmentLatch.countDown());
+
 		adapter.setMessageConverter(new BatchMessageConverter() {
 
 			@Override
@@ -480,12 +487,17 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment a
 		adapter.setErrorChannel(errors);
 		template.sendDefault(0, 1487694048607L, 1, "foo");
 		template.sendDefault(0, 1487694048608L, 1, "bar");
+
+		adapter.start();
+		ContainerTestUtils.waitForAssignment(container, 1);
+
 		Message<?> error = errors.receive(30000);
 		assertThat(error).isNotNull();
 		assertThat(error.getPayload()).isInstanceOf(ConversionException.class);
 		assertThat(((ConversionException) error.getPayload()).getMessage())
 				.contains("Failed to convert to message");
 		assertThat(((ConversionException) error.getPayload()).getRecords()).hasSize(2);
+		assertThat(restartAssignmentLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
 		adapter.stop();
 		pf.reset();