From 5f006dd43ec8bf6bcc96fb66dae190f500c348a2 Mon Sep 17 00:00:00 2001 From: streamnativebot Date: Tue, 13 Oct 2026 14:23:50 +0000 Subject: [PATCH 1/6] "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 dc84b2694..7c3b65bad 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 9ca7754eb8b37b6766eab8d454708bf1892a9e64 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 11:38:14 +0800 Subject: [PATCH 2/6] Trigger CI for branch-4.0.14.1 From 461866b5361a85383028b378f7e930aae30169ce Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 12:43:22 +0800 Subject: [PATCH 3/6] Remove duplicate Paho test dependency --- tests/pom.xml | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/tests/pom.xml b/tests/pom.xml index 7c3b65bad..8e6de32d4 100644 --- a/tests/pom.xml +++ b/tests/pom.xml @@ -81,12 +81,6 @@ ${hivemq.mqtt.client.version} test - - org.eclipse.paho - org.eclipse.paho.client.mqttv3 - 1.2.5 - test - @@ -118,4 +112,4 @@ https://repo.eclipse.org/content/repositories/paho-releases/ - \ No newline at end of file + From f5b47de110fb448cf424712052ed710532b0d7d3 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 12:58:34 +0800 Subject: [PATCH 4/6] Pin compatible dependency versions --- pom.xml | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 01c6c5288..fe7323415 100644 --- a/pom.xml +++ b/pom.xml @@ -58,7 +58,8 @@ 1.2.2 1.9.4 1.45.1 - 2.14.2 + 2.16.1 + 2.18.11 4.1.122.Final 17 @@ -130,6 +131,23 @@ + + com.fasterxml.jackson + jackson-bom + ${jackson.version} + pom + import + + + org.asynchttpclient + async-http-client + ${async-http-client.version} + + + org.asynchttpclient + async-http-client-netty-utils + ${async-http-client.version} + io.streamnative streamnative-bom From c88c42bd09b968b0a57baffc67c255ac3a385b37 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 13:02:23 +0800 Subject: [PATCH 5/6] Revert "Pin compatible dependency versions" This reverts commit f5b47de110fb448cf424712052ed710532b0d7d3. --- pom.xml | 20 +------------------- 1 file changed, 1 insertion(+), 19 deletions(-) diff --git a/pom.xml b/pom.xml index fe7323415..01c6c5288 100644 --- a/pom.xml +++ b/pom.xml @@ -58,8 +58,7 @@ 1.2.2 1.9.4 1.45.1 - 2.16.1 - 2.18.11 + 2.14.2 4.1.122.Final 17 @@ -131,23 +130,6 @@ - - com.fasterxml.jackson - jackson-bom - ${jackson.version} - pom - import - - - org.asynchttpclient - async-http-client - ${async-http-client.version} - - - org.asynchttpclient - async-http-client-netty-utils - ${async-http-client.version} - io.streamnative streamnative-bom From a1fc05ec10e951d2df29d416458e009fc1bb21b4 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 14 Oct 2026 13:05:55 +0800 Subject: [PATCH 6/6] 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");