INT-4457: Add logAndReply() terminal operator

JIRA: https://jira.spring.io/browse/INT-4457

When an `IntegrationFlow` is reply-based (we expect a reply in the
beginning) and `log()` (or `wireTap()`) is used in the end, we are
forced to add an empty `bridge()` in the end to ensure a `replyChannel`
header routing

* Fix `IntegrationFlowDefinition` to add `enrichHeaders()` in the end
to populate a `nullChannel` as a `replyChannel` header if that is missed
in the request message headers.
This way we cover both use-cases when we expect reply from the flow and
when it is used as a one-way scenario
* Improve a `HeaderEnricher` do not create a new `Message` if there are
no new headers to add/remove
* Remove a note from the `dsl.adoc` about now redundant `bridge()` after
`log()`
* Resolve TODO in the `.handle()` paragraph

* Restore anonymous class in the MessageHistoryIntegrationTests:
Mockito can't mock/spy lambdas because they are `final` classes

* Introduce `IntegrationFlowDefinition.logAndReply()` operator

* Fix logging message

Doc Polishing

src/reference/asciidoc/dsl.adoc

@@ -511,14 +511,31 @@ public IntegrationFlow integerFlow() {
 ----
 ====
 
-We also register a `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`.
-// TODO We don't show how to register a `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`.
+We also can register some `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`:
+
+====
+[source,java]
+----
+@Bean
+@IntegrationConverter
+public BytesToIntegerConverter bytesToIntegerConverter() {
+   return new BytesToIntegerConverter();
+}
+
+@Bean
+public IntegrationFlow integerFlow() {
+    return IntegrationFlows.from("input")
+             .handle(Integer.class, (p, h) -> p * 2)
+            .get();
+}
+----
+====
 
 [[java-dsl-log]]
 === Operator log()
 
 For convenience, to log the message journey through the Spring Integration flow (`<logging-channel-adapter>`), a `log()` operator is presented.
-Internally, it is represented by the `WireTap` `ChannelInterceptor` and `LoggingHandler` as subscriber.
+Internally, it is represented by the `WireTap` `ChannelInterceptor` with a `LoggingHandler` as its subscriber.
 It is responsible for logging the incoming message into the next endpoint or the current channel.
 The following example shows how to use `LoggingHandler`:
 
@@ -533,6 +550,10 @@ The following example shows how to use `LoggingHandler`:
 
 In the preceding example, an `id` header is logged at the `ERROR` level onto `test.category` only for messages that passed the filter and before routing.
 
+When this operator is used at the end of a flow, it is a one-way handler and the flow ends.
+To make it as a reply-producing flow, you can either use a simple `bridge()` after the `log()` or,  starting with version 5.1, you can use a `logAndReply()` operator instead.
+`logAndReply` can only be used at the end of a flow.
+
 [[java-dsl-wiretap]]
 === `MessageChannelSpec.wireTap()`
 
@@ -589,26 +610,6 @@ The following example does not have any channel declaration:
 
 In the preceding example (and any time no channel has been declared), an implicit `DirectChannel` is injected in the current position of the `IntegrationFlow` and used as an output channel for the currently configured `ServiceActivatingHandler` (from the `.handle()`, <<java-dsl-handle,described earlier>>).
 
-[IMPORTANT]
-====
-If `log()` or `wireTap()` are used in the end of the flow, they are considered to be one-way `MessageHandler` instances.
-If you expect the integration flow to return a reply, you should add a `bridge()` should to the end, after `log()` or `wireTap()`, as the following example shows:
-
-[source,java]
-----
-@Bean
-public IntegrationFlow sseFlow() {
-    return IntegrationFlows
-        .from(WebFlux.inboundGateway("/sse")
-            .requestMapping(m ->
-                m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
-        .handle((p, h) -> Flux.just("foo", "bar", "baz"))
-        .log(LoggingHandler.Level.WARN)
-        .bridge()
-        .get();
-}
-----
-====
 
 [[java-dsl-flows]]
 === Working With Message Flows

src/reference/asciidoc/whats-new.adoc

@@ -36,6 +36,9 @@ The following changes have been made in version 5.1:
 
 The `IntegrationFlowContext` is now an interface and `IntegrationFlowRegistration` is an inner interface of `IntegrationFlowContext`.
 
+A new `logAndReply()` operator has been introduced for convenience when you wish to log at the end of a flow for request-reply configurations.
+This avoid confusion with `log()` which is treated as a one-way end flow component.
+
 [[x5.1-dispatcher-exceptions]]
 ==== Dispatcher Exceptions