From ea4fe6d669e35b47b7f82522ef7fb5e07704aa81 Mon Sep 17 00:00:00 2001 From: streamnativebot Date: Wed, 14 Oct 2026 07:19:30 +0000 Subject: [PATCH 1/3] "Created by snbot" --- mqtt-broker/pom.xml | 2 +- mqtt-common/pom.xml | 2 +- mqtt-proxy/pom.xml | 2 +- pom.xml | 6 +++--- tests/pom.xml | 2 +- 5 files changed, 7 insertions(+), 7 deletions(-) diff --git a/mqtt-broker/pom.xml b/mqtt-broker/pom.xml index 2293b02ff..8fe393e34 100644 --- a/mqtt-broker/pom.xml +++ b/mqtt-broker/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 3.4.0-SNAPSHOT + 4.0.14.1 4.0.0 pulsar-protocol-handler-mqtt diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml index 707ea52b6..d0a184425 100644 --- a/mqtt-common/pom.xml +++ b/mqtt-common/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 3.4.0-SNAPSHOT + 4.0.14.1 4.0.0 pulsar-protocol-handler-mqtt-common diff --git a/mqtt-proxy/pom.xml b/mqtt-proxy/pom.xml index 47fdb6df7..852aba200 100644 --- a/mqtt-proxy/pom.xml +++ b/mqtt-proxy/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 3.4.0-SNAPSHOT + 4.0.14.1 4.0.0 pulsar-protocol-handler-mqtt-proxy diff --git a/pom.xml b/pom.xml index 5cfbdb037..01c6c5288 100644 --- a/pom.xml +++ b/pom.xml @@ -19,7 +19,7 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> io.streamnative.pulsar.handlers pulsar-protocol-handler-mqtt-parent - 3.4.0-SNAPSHOT + 4.0.14.1 StreamNative :: Pulsar Protocol Handler :: MoP Parent Parent for MQTT on Pulsar implemented using Pulsar Protocol Handler. @@ -51,8 +51,8 @@ 2.22.0 6.14.3 4.0.2 - 4.0.10.9 - 4.0.10.9 + 4.0.14.1 + 4.0.14.1 2.18.0 1.16 1.2.2 diff --git a/tests/pom.xml b/tests/pom.xml index ae2736668..8e6de32d4 100644 --- a/tests/pom.xml +++ b/tests/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 3.4.0-SNAPSHOT + 4.0.14.1 4.0.0 pulsar-protocol-handler-mqtt-tests From 9b499331a083ee202bd91925b1eb11c1eb56b742 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 13:05:55 +0800 Subject: [PATCH 2/3] Avoid Pulsar FieldParser for MQTT config parsing --- .../mqtt/common/utils/ConfigurationUtils.java | 158 +++++++++++++++++- 1 file changed, 156 insertions(+), 2 deletions(-) diff --git a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java index 461f6f249..c01c19260 100644 --- a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java +++ b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java @@ -16,8 +16,6 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static java.lang.String.format; -import static org.apache.pulsar.common.util.FieldParser.setEmptyValue; -import static org.apache.pulsar.common.util.FieldParser.value; import com.google.common.base.Joiner; import java.io.FileInputStream; import java.io.IOException; @@ -26,11 +24,14 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Properties; import java.util.Set; +import java.util.stream.Collectors; import java.util.stream.Stream; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.common.configuration.FieldContext; @@ -155,6 +156,159 @@ public static void update(Map properties, T obj) throws Ille }); } + private static Object value(String value, Field field) { + checkNotNull(field); + Type fieldType = field.getGenericType(); + if (fieldType instanceof ParameterizedType) { + Type[] actualTypes = ((ParameterizedType) fieldType).getActualTypeArguments(); + if (field.getType().equals(List.class)) { + return stringToList(value, getRawClass(actualTypes[0], field)); + } else if (field.getType().equals(Set.class)) { + return stringToSet(value, getRawClass(actualTypes[0], field)); + } else if (field.getType().equals(Map.class)) { + return stringToMap(value, getRawClass(actualTypes[0], field), getRawClass(actualTypes[1], field)); + } else if (field.getType().equals(Optional.class)) { + Type typeClazz = actualTypes[0]; + if (typeClazz instanceof ParameterizedType) { + throw new IllegalArgumentException(format("unsupported non-primitive Optional<%s> for %s", + typeClazz.getClass(), field.getName())); + } + return Optional.ofNullable(convert(value, getRawClass(typeClazz, field))); + } + throw new IllegalArgumentException(format("unsupported field-type %s for %s", + field.getType(), field.getName())); + } + return convert(value, field.getType()); + } + + private static void setEmptyValue(String value, Field field, T obj) + throws IllegalArgumentException, IllegalAccessException { + checkNotNull(field); + Type fieldType = field.getGenericType(); + if (fieldType instanceof ParameterizedType) { + if (field.getType().equals(List.class)) { + field.set(obj, new ArrayList<>()); + } else if (field.getType().equals(Set.class)) { + field.set(obj, new LinkedHashSet<>()); + } else if (field.getType().equals(Optional.class)) { + field.set(obj, Optional.empty()); + } else { + throw new IllegalArgumentException(format("unsupported field-type %s for %s", + field.getType(), field.getName())); + } + } else if (Number.class.isAssignableFrom(field.getType())) { + field.set(obj, null); + } + } + + private static Object convert(Object value, Class targetType) { + checkNotNull(targetType); + if (value == null) { + return null; + } + Class wrappedType = wrap(targetType); + if (wrappedType.isAssignableFrom(value.getClass())) { + return wrappedType.cast(value); + } + if (wrappedType.isEnum()) { + return stringToEnum(value.toString(), wrappedType); + } + if (wrappedType.equals(Integer.class)) { + return stringToInteger(value.toString()); + } else if (wrappedType.equals(Long.class)) { + return Long.valueOf(value.toString().trim()); + } else if (wrappedType.equals(Double.class)) { + return stringToDouble(value.toString()); + } else if (wrappedType.equals(Float.class)) { + return Float.valueOf(value.toString().trim()); + } else if (wrappedType.equals(Boolean.class)) { + return Boolean.valueOf(value.toString()); + } + throw new UnsupportedOperationException(format("Cannot convert from %s to %s. Requested converter does not" + + " exist.", value.getClass().getName(), wrappedType.getName())); + } + + private static Class getRawClass(Type type, Field field) { + if (type instanceof Class) { + return (Class) type; + } + throw new IllegalArgumentException(format("unsupported field-type %s for %s", type, field.getName())); + } + + private static Class wrap(Class clazz) { + if (clazz.equals(Integer.TYPE)) { + return Integer.class; + } else if (clazz.equals(Float.TYPE)) { + return Float.class; + } else if (clazz.equals(Double.TYPE)) { + return Double.class; + } else if (clazz.equals(Long.TYPE)) { + return Long.class; + } else if (clazz.equals(Boolean.TYPE)) { + return Boolean.class; + } + return clazz; + } + + private static Integer stringToInteger(String value) { + String trimmedValue = value.trim(); + if (StringUtils.isEmpty(trimmedValue)) { + return null; + } + return Integer.valueOf(trimmedValue); + } + + private static Double stringToDouble(String value) { + String trimmedValue = value.trim(); + if (StringUtils.isEmpty(trimmedValue)) { + return null; + } + return Double.valueOf(trimmedValue); + } + + private static List stringToList(String value, Class type) { + if (value == null) { + return null; + } + return Arrays.stream(value.trim().split(LISTENER_DEL)) + .map(item -> type.cast(convert(item.trim(), type))) + .collect(Collectors.toList()); + } + + private static Set stringToSet(String value, Class type) { + if (value == null) { + return null; + } + return Arrays.stream(value.trim().split(LISTENER_DEL)) + .map(item -> type.cast(convert(item.trim(), type))) + .collect(Collectors.toCollection(LinkedHashSet::new)); + } + + private static Map stringToMap(String value, Class keyType, Class valueType) { + if (value == null) { + return null; + } + return Arrays.stream(value.trim().split(LISTENER_DEL)) + .map(item -> item.trim().split("=", 2)) + .peek(parts -> checkArgument(parts.length == 2, + value + " map-value is not in correct format key1=value,key2=value2")) + .collect(Collectors.toMap( + parts -> keyType.cast(convert(parts[0].trim(), keyType)), + parts -> valueType.cast(convert(parts[1].trim(), valueType)), + (oldValue, newValue) -> newValue)); + } + + private static Object stringToEnum(String value, Class enumType) { + String trimmedValue = value.trim(); + for (Object enumConstant : enumType.getEnumConstants()) { + Enum enumValue = (Enum) enumConstant; + if (enumValue.toString().equals(trimmedValue) || enumValue.name().equals(trimmedValue)) { + return enumValue; + } + } + throw new RuntimeException(format("Invalid value '%s' for enum %s", value, enumType)); + } + public static int getListenerPort(String listener) { checkArgument(listener.matches(LISTENER_PATTERN), "listener not match pattern"); From 2b9b6de028b8d7857210ebaf2a4fe419784d4541 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 16:10:29 +0800 Subject: [PATCH 3/3] Avoid offload policy parsing in MQTT tests --- .../pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java index 47a689fa2..28ca3d293 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java @@ -142,6 +142,7 @@ protected MQTTCommonConfiguration initConfig() throws Exception{ mqtt.setDefaultNumberOfNamespaceBundles(4); mqtt.setZookeeperServers("localhost:2181"); mqtt.setConfigurationStoreServers("localhost:3181"); + mqtt.setTopicLevelPoliciesEnabled(false); mqtt.setAuthenticationEnabled(false); mqtt.setAuthorizationEnabled(false); @@ -390,6 +391,8 @@ protected void startBroker(MQTTCommonConfiguration conf) throws Exception { mqttBrokerPort, mqttBrokerTlsPort, mqttBrokerTlsPskPort, mqttProxyPort, mqttProxyTlsPort, mqttProxyTlsPskPort); ConfigurationUtils.extractFieldToProperties(conf); + conf.getProperties().remove("managedLedgerDataReadPriority"); + conf.getProperties().remove("managedLedgerOffloadedReadPriority"); setTLSConf(conf); mockZooKeeper.setSessionId(pulsarServiceList.size() + 1); this.pulsarServiceList.add(doStartBroker(conf));