[fix][broker] Avoid IllegalStateException when marker_type field is not set in publishing (#24087)

lhotari · web-flow · commit 3e6f7deb0afe · 2025-03-17T06:51:52.000-07:00
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -363,7 +363,7 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
             // Message is coming from replication, we need to use the replication's producer name, ledger id and entry
             // id for the purpose of deduplication.
             MessageMetadata md = Commands.peekMessageMetadata(headersAndPayload, "Check-Deduplicate", -1);
-            if (Markers.isReplicationMarker(md.getMarkerType())) {
+            if (md != null && md.hasMarkerType() && Markers.isReplicationMarker(md.getMarkerType())) {
                 publishContext.setProperty(MSG_PROP_IS_REPL_MARKER, "");
             }
             return;

Original file line number	Diff line number	Diff line change
`@@ -363,7 +363,7 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header`
`363`	`363`	`// Message is coming from replication, we need to use the replication's producer name, ledger id and entry`
`364`	`364`	`// id for the purpose of deduplication.`
`365`	`365`	`MessageMetadata md = Commands.peekMessageMetadata(headersAndPayload, "Check-Deduplicate", -1);`
`366`		`- if (Markers.isReplicationMarker(md.getMarkerType())) {`
	`366`	`+ if (md != null && md.hasMarkerType() && Markers.isReplicationMarker(md.getMarkerType())) {`
`367`	`367`	`publishContext.setProperty(MSG_PROP_IS_REPL_MARKER, "");`
`368`	`368`	`}`
`369`	`369`	`return;`