fix(rocketmq): log warning when FLAG or DELAY header is not numeric

RocketMQMessageConverterSupport#getAndWrapMessage swallowed any
Integer.parseInt failure on the MQ_FLAG and DELAY_TIME_LEVEL message
headers with catch (Exception ignored) {}, silently falling back to
flag=0 and delayLevel=0. A producer that set DELAY to a misconfigured
non-numeric value (e.g. via a SpEL expression that produced a String
instead of an int) would have its delayed delivery silently turn into
an immediate send with no log line and no way to know the header was
malformed.

Parse each header independently in a small parseHeaderAsIntOrDefault
helper that warns, naming the offending header and echoing its raw
value, when parsing fails. Parsing both headers in a single shared try
block had two problems: a valid value on the header parsed first (e.g.
a legitimate DELAY) was reported in the warning as having fallen back
to 0 even though it was actually applied, and an exception on the
header parsed first left the other, well-formed header unparsed and
forced it to 0 as well. Independent parsing keeps every well-formed
value, and the warning now names only the header that actually failed.
Only the exception message is logged (not the full stack trace) to
avoid flooding the logs when many messages carry the same bad header,
and the catch is narrowed from Exception to NumberFormatException.

Adds regression tests covering both the fallback behaviour (non-numeric
DELAY and FLAG each stay at 0, conversion never propagates a parse
exception) and the behaviour introduced here: the warning is emitted
for a malformed header (asserted with OutputCaptureExtension), a valid
delay level survives a malformed flag on the same message, and a valid
flag survives a malformed delay level.

Author: daguimu

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/support/RocketMQMessageConverterSupport.java

@@ -26,6 +26,8 @@
 import com.alibaba.cloud.stream.binder.rocketmq.custom.RocketMQBeanContainerCache;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -40,6 +42,9 @@
  */
 public final class RocketMQMessageConverterSupport {
 
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQMessageConverterSupport.class);
+
 	private RocketMQMessageConverterSupport() {
 	}
 
@@ -153,20 +158,15 @@ private static org.apache.rocketmq.common.message.Message getAndWrapMessage(
 			}
 			Object flagObj = headers.getOrDefault(Headers.FLAG,
 					headers.get(toRocketHeaderKey(Headers.FLAG)));
-			int flag = 0;
-			int delayLevel = 0;
-			try {
-				flagObj = flagObj == null ? 0 : flagObj;
-				Object delayLevelObj = headers.getOrDefault(
-						RocketMQConst.PROPERTY_DELAY_TIME_LEVEL,
-						headers.get(toRocketHeaderKey(
-								RocketMQConst.PROPERTY_DELAY_TIME_LEVEL)));
-				delayLevelObj = delayLevelObj == null ? 0 : delayLevelObj;
-				delayLevel = Integer.parseInt(String.valueOf(delayLevelObj));
-				flag = Integer.parseInt(String.valueOf(flagObj));
-			}
-			catch (Exception ignored) {
-			}
+			Object delayLevelObj = headers.getOrDefault(
+					RocketMQConst.PROPERTY_DELAY_TIME_LEVEL,
+					headers.get(toRocketHeaderKey(
+							RocketMQConst.PROPERTY_DELAY_TIME_LEVEL)));
+			// Parse each header independently so a malformed value in one never
+			// forces the other, well-formed header to fall back to 0.
+			int flag = parseHeaderAsIntOrDefault(Headers.FLAG, flagObj);
+			int delayLevel = parseHeaderAsIntOrDefault(
+					RocketMQConst.PROPERTY_DELAY_TIME_LEVEL, delayLevelObj);
 			if (delayLevel > 0) {
 				rocketMsg.setDelayTimeLevel(delayLevel);
 			}
@@ -191,4 +191,30 @@ private static org.apache.rocketmq.common.message.Message getAndWrapMessage(
 		return rocketMsg;
 	}
 
+	/**
+	 * Parses a message header value as an {@code int}, falling back to {@code 0}
+	 * when the header is absent or not a valid integer. Each header is parsed on
+	 * its own so a malformed value in one header never drags a well-formed value
+	 * in another header down to the fallback, and the warning names only the
+	 * header that actually failed. Only the exception message is logged (no stack
+	 * trace) to avoid flooding the logs when many messages carry the same bad
+	 * header.
+	 * @param headerName the header name, used only for the warning message
+	 * @param rawValue the raw header value, may be {@code null}
+	 * @return the parsed value, or {@code 0} when absent or non-numeric
+	 */
+	private static int parseHeaderAsIntOrDefault(String headerName, Object rawValue) {
+		if (rawValue == null) {
+			return 0;
+		}
+		try {
+			return Integer.parseInt(String.valueOf(rawValue));
+		}
+		catch (NumberFormatException e) {
+			log.warn("Non-numeric '{}' header value [{}]; falling back to 0: {}",
+					headerName, rawValue, e.getMessage());
+			return 0;
+		}
+	}
+
 }

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/test/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQMessageConverterSupportTest.java

@@ -16,18 +16,24 @@
 
 package com.alibaba.cloud.stream.binder.rocketmq;
 
+import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
 import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQMessageConverterSupport;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 /**
  * @author Sorie
  */
+@ExtendWith(OutputCaptureExtension.class)
 public class RocketMQMessageConverterSupportTest {
 
 	@Test
@@ -44,4 +50,97 @@ public void convertMessage2MQBlankHeaderTest() {
 		assertThat(testProp).isNull();
 		assertThat(tagProp).isEqualTo("a");
 	}
+
+	@Test
+	public void nonNumericDelayTimeLevelHeaderFallsBackToZero() {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.PROPERTY_DELAY_TIME_LEVEL, "not-a-number")
+				.build();
+
+		org.apache.rocketmq.common.message.Message rkmqMsg =
+				RocketMQMessageConverterSupport.convertMessage2MQ("topic", message);
+
+		assertThat(rkmqMsg.getDelayTimeLevel()).isEqualTo(0);
+		assertThat(rkmqMsg.getFlag()).isEqualTo(0);
+	}
+
+	@Test
+	public void nonNumericFlagHeaderFallsBackToZero() {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.Headers.FLAG, "not-a-number")
+				.build();
+
+		org.apache.rocketmq.common.message.Message rkmqMsg =
+				RocketMQMessageConverterSupport.convertMessage2MQ("topic", message);
+
+		assertThat(rkmqMsg.getFlag()).isEqualTo(0);
+	}
+
+	@Test
+	public void invalidNumericHeaderDoesNotPropagateException() {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.PROPERTY_DELAY_TIME_LEVEL, "x")
+				.setHeader(RocketMQConst.Headers.FLAG, "y")
+				.build();
+
+		assertThatCode(() ->
+				RocketMQMessageConverterSupport.convertMessage2MQ("topic", message))
+				.doesNotThrowAnyException();
+	}
+
+	@Test
+	public void nonNumericFlagHeaderLogsWarning(CapturedOutput output) {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.Headers.FLAG, "flag-not-a-number")
+				.build();
+
+		RocketMQMessageConverterSupport.convertMessage2MQ("topic", message);
+
+		// The malformed header must no longer be swallowed silently: the warning
+		// names the offending header and echoes its raw value.
+		assertThat(output).contains(RocketMQConst.Headers.FLAG)
+				.contains("flag-not-a-number");
+	}
+
+	@Test
+	public void validDelayLevelWithNonNumericFlagIsPreservedAndWarnsOnlyFlag(
+			CapturedOutput output) {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.PROPERTY_DELAY_TIME_LEVEL, "3")
+				.setHeader(RocketMQConst.Headers.FLAG, "flag-bad")
+				.build();
+
+		org.apache.rocketmq.common.message.Message rkmqMsg =
+				RocketMQMessageConverterSupport.convertMessage2MQ("topic", message);
+
+		// A valid delay level must survive a malformed flag on the same message;
+		// the two headers are parsed independently.
+		assertThat(rkmqMsg.getDelayTimeLevel()).isEqualTo(3);
+		assertThat(rkmqMsg.getFlag()).isEqualTo(0);
+		// The warning names the flag header only; the valid delay level must not
+		// be reported as having fallen back.
+		assertThat(output).contains("flag-bad")
+				.doesNotContain("'" + RocketMQConst.PROPERTY_DELAY_TIME_LEVEL + "'");
+	}
+
+	@Test
+	public void validFlagWithNonNumericDelayLevelIsPreservedAndWarnsOnlyDelayLevel(
+			CapturedOutput output) {
+		Message<String> message = MessageBuilder.withPayload("msg")
+				.setHeader(RocketMQConst.PROPERTY_DELAY_TIME_LEVEL, "delay-bad")
+				.setHeader(RocketMQConst.Headers.FLAG, "7")
+				.build();
+
+		org.apache.rocketmq.common.message.Message rkmqMsg =
+				RocketMQMessageConverterSupport.convertMessage2MQ("topic", message);
+
+		// A valid flag must survive a malformed delay level on the same message;
+		// the two headers are parsed independently.
+		assertThat(rkmqMsg.getFlag()).isEqualTo(7);
+		assertThat(rkmqMsg.getDelayTimeLevel()).isEqualTo(0);
+		// The warning names the delay-level header only; the valid flag must not
+		// be reported as having fallen back.
+		assertThat(output).contains("delay-bad")
+				.doesNotContain("'" + RocketMQConst.Headers.FLAG + "'");
+	}
 }