Fix DSL for inner bean names generation (#8639)

The `IntegrationFlowBeanPostProcessor.processIntegrationComponentSpec()`
uses a wrong `generateBeanName()` for component to register making the
provided `id` as a prefix

* Use `generateBeanName(Object instance, String prefix, @nullable String fallbackId, boolean useFlowIdAsPrefix)`
instead to properly "fallback" to the provided name

**Cherry-pick to `6.1.x`**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java

@@ -356,7 +356,7 @@ private void processIntegrationComponentSpec(String beanName, IntegrationCompone
 						.filter(component -> noBeanPresentForComponent(component.getKey(), beanName))
 						.forEach(component ->
 								registerComponent(component.getKey(),
-										generateBeanName(component.getKey(), component.getValue())));
+										generateBeanName(component.getKey(), beanName, component.getValue(), false)));
 
 			}
 		}

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java

@@ -244,8 +244,10 @@ void testGateways() throws Exception {
 	}
 
 	@Test
-	void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel,
-			@Autowired PollableKafkaChannel topic9Channel) {
+	void channels(@Qualifier("topic6Channel") MessageChannel topic6Channel,
+			@Qualifier("topic8Channel") PollableKafkaChannel topic8Channel,
+			@Qualifier("topic9Channel") PollableKafkaChannel topic9Channel) {
+
 		topic6Channel.send(new GenericMessage<>("foo"));
 		Message<?> received = topic8Channel.receive();
 		assertThat(received)
@@ -344,7 +346,8 @@ public PollableChannel futuresChannel() {
 		}
 
 		@Bean
-		public IntegrationFlow sendToKafkaFlow() {
+		public IntegrationFlow sendToKafkaFlow(
+				KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandlerTopic2) {
 			return f -> f
 					.<String>split(p -> Stream.generate(() -> p).limit(101).iterator(), null)
 					.enrichHeaders(h -> h.header(KafkaIntegrationHeaders.FUTURE_TOKEN, "foo"))
@@ -353,17 +356,15 @@ public IntegrationFlow sendToKafkaFlow() {
 									kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
 											.timestampExpression("T(Long).valueOf('1487694048633')"),
 									e -> e.id("kafkaProducer1")))
-							.subscribe(sf -> sf.handle(
-									kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
-											.flush(msg -> true)
-											.timestamp(m -> 1487694048644L),
-									e -> e.id("kafkaProducer2")))
+							.subscribe(sf -> sf.handle(kafkaMessageHandlerTopic2, e -> e.id("kafkaProducer2")))
 					);
 		}
 
 		@Bean
-		public DefaultKafkaHeaderMapper mapper() {
-			return new DefaultKafkaHeaderMapper();
+		public KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandlerTopic2() {
+			return kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
+					.flush(msg -> true)
+					.timestamp(m -> 1487694048644L);
 		}
 
 		private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
@@ -382,6 +383,11 @@ public DefaultKafkaHeaderMapper mapper() {
 					.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
 		}
 
+		@Bean
+		public DefaultKafkaHeaderMapper mapper() {
+			return new DefaultKafkaHeaderMapper();
+		}
+
 
 		@Bean
 		public IntegrationFlow sourceFlow() {
@@ -427,7 +433,7 @@ public KafkaMessageSource<Integer, String> channelSource(ConsumerFactory<Integer
 		@Bean
 		public IntegrationFlow channels(KafkaTemplate<Integer, String> template,
 				ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory,
-				KafkaMessageSource<?, ?> channelSource,
+				@Qualifier("channelSource") KafkaMessageSource<?, ?> channelSource,
 				PublishSubscribeKafkaChannel publishSubscribeKafkaChannel) {
 
 			return IntegrationFlow.from(topic6Channel(template, containerFactory))