rdhabalia closed pull request #1509: Fix init of Input CustomSerDe of function
URL: https://github.com/apache/incubator-pulsar/pull/1509
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-admin/pom.xml b/pulsar-client-admin/pom.xml
index e0665b1179..fec7b91212 100644
--- a/pulsar-client-admin/pom.xml
+++ b/pulsar-client-admin/pom.xml
@@ -46,15 +46,15 @@
<version>${project.version}</version>
</dependency>
- <dependency>
+ <dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-functions-proto-shaded</artifactId>
<version>${project.version}</version>
<exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
</exclusions>
</dependency>
diff --git
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index 510608f8e4..a4bf07ae4e 100644
---
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -723,7 +723,10 @@ private void setupSerDe(Class<?>[] typeArgs, ClassLoader
clsLoader) {
this.inputSerDe = new HashMap<>();
instanceConfig.getFunctionConfig().getCustomSerdeInputsMap().forEach((k, v) ->
this.inputSerDe.put(k, initializeSerDe(v, clsLoader, typeArgs, true)));
for (String topicName :
instanceConfig.getFunctionConfig().getInputsList()) {
- this.inputSerDe.put(topicName, initializeDefaultSerDe(typeArgs,
true));
+ // initialize Default-SerDe if custom-SerDe is not attached to the
topic
+ if (!inputSerDe.containsKey(topicName)) {
+ this.inputSerDe.put(topicName,
initializeDefaultSerDe(typeArgs, true));
+ }
}
if (Void.class.equals(typeArgs[0])) {
diff --git
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
index 1e9781c35c..5963bf31ac 100644
---
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
+++
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
@@ -21,6 +21,8 @@
import lombok.Getter;
import lombok.Setter;
import net.jodah.typetools.TypeResolver;
+
+import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.api.SerDe;
@@ -47,13 +49,14 @@ public Integer deserialize(byte[] input) {
}
}
- private static InstanceConfig createInstanceConfig(boolean addCustom,
String outputSerde) {
+ private static InstanceConfig createInstanceConfig(boolean addCustom,
String outputSerde, String inputSerde) {
FunctionConfig.Builder functionConfigBuilder =
FunctionConfig.newBuilder();
if (!addCustom) {
functionConfigBuilder.addInputs("TEST");
} else {
- functionConfigBuilder.putCustomSerdeInputs("TEST",
IntegerSerDe.class.getName());
+ functionConfigBuilder.putCustomSerdeInputs("TEST", inputSerde);
}
+ functionConfigBuilder.addInputs("TEST");
if (outputSerde != null) {
functionConfigBuilder.setOutputSerdeClassName(outputSerde);
}
@@ -63,8 +66,8 @@ private static InstanceConfig createInstanceConfig(boolean
addCustom, String out
return instanceConfig;
}
- private JavaInstanceRunnable createRunnable(boolean addCustom, String
outputSerde) throws Exception {
- InstanceConfig config = createInstanceConfig(addCustom, outputSerde);
+ private JavaInstanceRunnable createRunnable(boolean addCustom, String
outputSerde, String inputSerde) throws Exception {
+ InstanceConfig config = createInstanceConfig(addCustom, outputSerde,
inputSerde);
JavaInstanceRunnable javaInstanceRunnable = new JavaInstanceRunnable(
config, null, null, null, null);
return javaInstanceRunnable;
@@ -122,7 +125,7 @@ public Void process(String input, Context context) throws
Exception {
@Test
public void testVoidInputClasses() {
try {
- JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName());
+ JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName(), IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
VoidInputHandler pulsarFunction = new VoidInputHandler();
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
@@ -142,7 +145,7 @@ public void testVoidInputClasses() {
@Test
public void testVoidOutputClasses() {
try {
- JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName());
+ JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName(), IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
VoidOutputHandler pulsarFunction = new VoidOutputHandler();
@@ -159,7 +162,7 @@ public void testVoidOutputClasses() {
@Test
public void testInconsistentInputType() {
try {
- JavaInstanceRunnable runnable = createRunnable(true,
DefaultSerDe.class.getName());
+ JavaInstanceRunnable runnable = createRunnable(true,
DefaultSerDe.class.getName(), IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
Function function = (Function<String, String>) (input, context) ->
input + "-lambda";
@@ -179,7 +182,7 @@ public void testInconsistentInputType() {
@Test
public void testDefaultSerDe() {
try {
- JavaInstanceRunnable runnable = createRunnable(false, null);
+ JavaInstanceRunnable runnable = createRunnable(false, null,
IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
Function function = (Function<String, String>) (input, context) ->
input + "-lambda";
@@ -198,7 +201,7 @@ public void testDefaultSerDe() {
@Test
public void testExplicitDefaultSerDe() {
try {
- JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName());
+ JavaInstanceRunnable runnable = createRunnable(false,
DefaultSerDe.class.getName(), IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
Function function = (Function<String, String>) (input, context) ->
input + "-lambda";
@@ -209,13 +212,32 @@ public void testExplicitDefaultSerDe() {
}
}
+ /**
+ * Verify that Explicit setting of Non-Default Serializer works fine.
+ */
+ @Test
+ public void testExplicitNonDefaultSerDe() {
+ try {
+ JavaInstanceRunnable runnable = createRunnable(true,
DefaultSerDe.class.getName(),
+ CustomTestDataSerDe.class.getName());
+ Method method = makeAccessible(runnable);
+ ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
+ Function function = (Function<TestData, String>) (input, context)
-> input + "-lambda";
+ Class<?>[] typeArgs =
TypeResolver.resolveRawArguments(Function.class, function.getClass());
+ method.invoke(runnable, typeArgs, clsLoader);
+ } catch (Exception ex) {
+ assertTrue(false);
+ }
+ }
+
/**
* Verify that function output type should be consistent with output serde
type.
*/
@Test
public void testInconsistentOutputType() {
try {
- JavaInstanceRunnable runnable = createRunnable(false,
IntegerSerDe.class.getName());
+ JavaInstanceRunnable runnable = createRunnable(false,
IntegerSerDe.class.getName(),
+ IntegerSerDe.class.getName());
Method method = makeAccessible(runnable);
ClassLoader clsLoader =
Thread.currentThread().getContextClassLoader();
Function function = (Function<String, String>) (input, context) ->
input + "-lambda";
@@ -229,5 +251,29 @@ public void testInconsistentOutputType() {
}
}
+ static class CustomTestDataSerDe implements SerDe<TestData> {
+ @Override
+ public TestData deserialize(byte[] input) {
+ try {
+ return ObjectMapperFactory.getThreadLocal().readValue(input,
TestData.class);
+ } catch (Exception e) {
+ String data = new String(input);
+ throw new IllegalArgumentException("invalid topic data " +
data, e);
+ }
+ }
+
+ @Override
+ public byte[] serialize(TestData input) {
+ try {
+ return
ObjectMapperFactory.getThreadLocal().writeValueAsBytes(input);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("invalid topic data " +
input, e);
+ }
+ }
+ }
+
+ static class TestData {
+ public String test;
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
