1717
1818import static java .nio .charset .StandardCharsets .UTF_8 ;
1919
20+ import java .nio .ByteBuffer ;
21+ import java .nio .ByteOrder ;
2022import java .util .HashMap ;
2123import javax .jms .BytesMessage ;
2224import javax .jms .JMSContext ;
3234/**
3335 * Builds Kafka Connect SourceRecords from messages. It parses the bytes of the payload of JMS
3436 * BytesMessage and TextMessage as JSON and creates a SourceRecord with a null schema.
37+ *
38+ * When messageBodyJms is false, this builder can handle MQ messages with RFH2 headers by
39+ * automatically detecting and skipping them to extract the JSON payload.
3540 */
3641public class JsonRecordBuilder extends BaseRecordBuilder {
3742 private static final Logger log = LoggerFactory .getLogger (JsonRecordBuilder .class );
3843
3944 private JsonConverter converter ;
4045
46+ // RFH2 header constants
47+ private static final String RFH2_STRUC_ID = "RFH " ;
48+ private static final int RFH2_STRUCT_ID_LENGTH = 4 ;
49+ private static final int RFH2_STRUC_LENGTH_OFFSET = 8 ;
50+ private static final int RFH2_MIN_HEADER_SIZE = 36 ;
51+
4152 public JsonRecordBuilder () {
4253 log .info ("Building records using com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder" );
4354 converter = new JsonConverter ();
@@ -65,7 +76,7 @@ public JsonRecordBuilder() {
6576 @ Override
6677 public SchemaAndValue getValue (final JMSContext context , final String topic , final boolean messageBodyJms ,
6778 final Message message ) throws JMSException {
68- final byte [] payload ;
79+ byte [] payload ;
6980
7081 if (message instanceof BytesMessage ) {
7182 payload = message .getBody (byte [].class );
@@ -77,6 +88,84 @@ public SchemaAndValue getValue(final JMSContext context, final String topic, fin
7788 throw new RecordBuilderException ("Unsupported JMS message type" );
7889 }
7990
91+ // When messageBodyJms is false, the message may contain RFH2 headers that need to be skipped
92+ if (!messageBodyJms ) {
93+ payload = stripRFH2Header (payload );
94+ }
8095 return converter .toConnectData (topic , payload );
8196 }
97+
98+ /**
99+ * Skips RFH2 (Rules and Formatting Header version 2) if present in the payload.
100+ * RFH2 headers are used by IBM MQ to carry additional message properties and metadata.
101+ *
102+ * When messageBodyJms is false (WMQ_MESSAGE_BODY_MQ), JMS does not automatically
103+ * strip RFH2 headers, so we need to parse and skip them manually.
104+ *
105+ * RFH2 structure:
106+ * - StrucId (4 bytes): "RFH " (with trailing space)
107+ * - Version (4 bytes): Version number (typically 2)
108+ * - StrucLength (4 bytes): Total length of RFH2 header including all folders
109+ * - Encoding (4 bytes): Numeric encoding
110+ * - CodedCharSetId (4 bytes): Character set identifier
111+ * - Format (8 bytes): Format name of data following the header
112+ * - Flags (4 bytes): Flags
113+ * - NameValueCCSID (4 bytes): CCSID of name-value data
114+ * - Variable length folders containing name-value pairs
115+ *
116+ * Inspired from https://github.com/CommunityHiQ/Frends.Community.IBMMQ/blob/master/Frends.Community.IBMMQ/Helpers/IBMMQHelpers.cs
117+ *
118+ * @param payload the original message payload
119+ * @return the payload with RFH2 header removed if present, otherwise the original payload
120+ */
121+ private byte [] stripRFH2Header (final byte [] payload ) {
122+ if (payload == null || payload .length < RFH2_MIN_HEADER_SIZE ) {
123+ return payload ;
124+ }
125+
126+ // Check if the message starts with RFH2 structure ID
127+ final String strucId = new String (payload , 0 , RFH2_STRUCT_ID_LENGTH , UTF_8 );
128+ if (!RFH2_STRUC_ID .equals (strucId )) {
129+ log .debug ("No RFH2 header detected" );
130+ return payload ;
131+ }
132+
133+ try {
134+ // Read version to detect endianness
135+ final ByteBuffer buffer = ByteBuffer .wrap (payload );
136+ buffer .order (ByteOrder .LITTLE_ENDIAN ); // Default to little-endian
137+ buffer .position (RFH2_STRUCT_ID_LENGTH ); // Skip StrucId (4 bytes)
138+ int version = buffer .getInt ();
139+
140+ // Detect endianness: if version is not 1 or 2, it's likely big-endian
141+ if (version > 2 || version < 1 ) {
142+ version = Integer .reverseBytes (version );
143+ buffer .order (ByteOrder .BIG_ENDIAN );
144+ log .debug ("Detected big-endian RFH2 header" );
145+ } else {
146+ log .debug ("Detected little-endian RFH2 header" );
147+ }
148+
149+ // Read the RFH2 structure length, Skip StrucId (4 bytes) and Version (4 bytes)
150+ buffer .position (RFH2_STRUC_LENGTH_OFFSET );
151+ final int strucLength = buffer .getInt ();
152+
153+ if (strucLength < RFH2_MIN_HEADER_SIZE || strucLength > payload .length ) {
154+ log .warn ("Invalid RFH2 structure length: {}. Treating entire payload as message." , strucLength );
155+ return payload ;
156+ }
157+
158+ log .debug ("RFH2 header detected (version: {}, length: {} bytes). Stripping header." , version , strucLength );
159+
160+ // Extract the actual message payload after the RFH2 header
161+ final byte [] actualPayload = new byte [payload .length - strucLength ];
162+ System .arraycopy (payload , strucLength , actualPayload , 0 , actualPayload .length );
163+
164+ return actualPayload ;
165+
166+ } catch (final Exception e ) {
167+ log .error ("Error parsing RFH2 header: {}. Returning original payload." , e .getMessage ());
168+ return payload ;
169+ }
170+ }
82171}
0 commit comments