[jira] [Created] (FLINK-22584) Use protobuf-shaded in StateFun core.

[Shang Yuanchun (Jira)]

Igal Shilman created FLINK-22584:
------------------------------------

             Summary: Use protobuf-shaded in StateFun core.
                 Key: FLINK-22584
                 URL: https://issues.apache.org/jira/browse/FLINK-22584
             Project: Flink
          Issue Type: Improvement
          Components: Stateful Functions
            Reporter: Igal Shilman


We have *statefun-protobuf-shaded* module, that is used by the remote java sdk,

we can use it to shade protobuf internally.

The major hurdle we need to overcome is that, in embedded functions, we have to be able to accept instances of protobuf generated messages by the user.

For example:
{code:java}
UserProfile userProfile = UserProfile.newBilder().build();
context.send(..., userProfile) {code}
If we will simply use the shaded Protobuf version, we will get immediately a class cast exception.

One way to overcome this is to use reflection and find the well known methods on the generated classes and call toBytes() / parseFrom() reflectively.

This however will cause a significant slow down, even by using MethodHandles.
A small experiment that I've previously done with ByteBuddy mitigates this, by generating
accessors, in pre-flight:

{code:java}
package org.apache.flink.statefun.flink.common.protobuf.serde;

import static net.bytebuddy.matcher.ElementMatchers.named;import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.DynamicType;
import net.bytebuddy.implementation.FixedValue;
import net.bytebuddy.implementation.MethodCall;
import net.bytebuddy.implementation.bytecode.assign.Assigner;final class ReflectiveProtobufSerde {  @SuppressWarnings({"unchecked", "rawtypes"})
  static <M> ProtobufSerde<M> ofProtobufGeneratedType(Class<M> type) {
    try {
      DynamicType.Unloaded<ProtobufSerde> unloaded = configureByteBuddy(type);      Class<? extends ProtobufSerde> writer = unloaded.load(type.getClassLoader()).getLoaded();      return (ProtobufSerde<M>) writer.getDeclaredConstructor().newInstance();
    } catch (Throwable e) {
      throw new IllegalArgumentException();
    }
  }  @SuppressWarnings("rawtypes")
  private static DynamicType.Unloaded<ProtobufSerde> configureByteBuddy(Class<?> type)
      throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
    Method writeToMethod = type.getMethod("writeTo", OutputStream.class);
    Method parseFromMethod = type.getMethod("parseFrom", InputStream.class);
    Method getSerializedSizeMethod = type.getMethod("getSerializedSize");    // get the message full name
    Method getDescriptorMethod = type.getMethod("getDescriptor");
    Object descriptor = getDescriptorMethod.invoke(null);
    Method getFullNameMethod = descriptor.getClass().getMethod("getFullName");
    String messageFullName = (String) getFullNameMethod.invoke(descriptor);    return new ByteBuddy()
        .subclass(ProtobufSerde.class)
        .typeVariable("M", type)
        .method(named("writeTo"))
        .intercept(
            MethodCall.invoke(writeToMethod)
                .onArgument(0)
                .withArgument(1)
                .withAssigner(Assigner.DEFAULT, Assigner.Typing.DYNAMIC))
        .method(named("parseFrom"))
        .intercept(MethodCall.invoke(parseFromMethod).withArgument(0))
        .method(named("getSerializedSize"))
        .intercept(
            MethodCall.invoke(getSerializedSizeMethod)
                .onArgument(0)
                .withAssigner(Assigner.DEFAULT, Assigner.Typing.DYNAMIC))
        .method(named("getMessageFullName"))
        .intercept(FixedValue.value(messageFullName))
        .make();
  }
}
 {code}
 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)