Add tracing headers to be mapped to string (#3286)

Fixed: #3286

For more context, see this Spring Cloud Stream issue: spring-cloud/spring-cloud-stream#2905

(cherry picked from commit 12ed737)

Author: artembilan

spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

@@ -49,14 +49,25 @@
  */
 public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 
+	private static final String[] DEFAULT_RAW_MAPPED_HEADERS = {
+			KafkaHeaders.LISTENER_INFO,
+			"b3",
+			"traceparent",
+			"X-B3-TraceId",
+			"X-B3-SpanId",
+			"X-B3-Sampled",
+			"X-B3-Flags"};
+
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
 	private final List<HeaderMatcher> matchers = new ArrayList<>();
 
 	private final Map<String, Boolean> rawMappedHeaders = new HashMap<>();
 
 	{
-		this.rawMappedHeaders.put(KafkaHeaders.LISTENER_INFO, true);
+		for (String defaultRawMappedHeader : DEFAULT_RAW_MAPPED_HEADERS) {
+			this.rawMappedHeaders.put(defaultRawMappedHeader, true);
+		}
 	}
 
 	private final boolean outbound;
@@ -153,14 +164,15 @@ public void setCharset(Charset charset) {
 	 * {@code byte[]} for outbound). Inbound headers that match will be mapped as
 	 * {@code byte[]} unless the corresponding boolean in the map value is true,
 	 * in which case it will be mapped as a String.
+	 * Headers mapped by default are: {@code kafka_listenerInfo}, {@code b3}, {@code traceparent},
+	 * {@code X-B3-TraceId}, {@code X-B3-SpanId}, {@code X-B3-Sampled} and {@code X-B3-Flags}.
 	 * @param rawMappedHeaders the header names to not convert and
 	 * @since 2.2.5
 	 * @see #setCharset(Charset)
 	 * @see #setMapAllStringsOut(boolean)
 	 */
 	public void setRawMappedHeaders(Map<String, Boolean> rawMappedHeaders) {
 		if (!ObjectUtils.isEmpty(rawMappedHeaders)) {
-			this.rawMappedHeaders.clear();
 			this.rawMappedHeaders.putAll(rawMappedHeaders);
 		}
 	}