@Test public void testCalculateSubjectTypeForFunction() { FunctionDetails.Builder builder = FunctionDetails.newBuilder(); builder.setSource(Function.SourceSpec.newBuilder().putInputSpecs("topic", Function.ConsumerSpec.newBuilder().build()).build()); assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.FUNCTION); builder.setComponentType(FunctionDetails.ComponentType.SOURCE); assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.SOURCE); builder.setComponentType(FunctionDetails.ComponentType.SINK); assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.SINK); }
public Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails) { if (functionDetails.getComponentType() != Function.FunctionDetails.ComponentType.UNKNOWN) { return functionDetails.getComponentType(); } Function.SourceSpec sourceSpec = functionDetails.getSource(); Function.SinkSpec sinkSpec = functionDetails.getSink(); if (sourceSpec.getInputSpecsCount() == 0) { return Function.FunctionDetails.ComponentType.SOURCE; } if (!isEmpty(sinkSpec.getBuiltin())) { return Function.FunctionDetails.ComponentType.SINK; } if (isEmpty(sinkSpec.getClassName()) || sinkSpec.getClassName().equals(PulsarSink.class.getName())) { return Function.FunctionDetails.ComponentType.FUNCTION; } return Function.FunctionDetails.ComponentType.SINK; }
InstanceUtils { public Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails) { if (functionDetails.getComponentType() != Function.FunctionDetails.ComponentType.UNKNOWN) { return functionDetails.getComponentType(); } Function.SourceSpec sourceSpec = functionDetails.getSource(); Function.SinkSpec sinkSpec = functionDetails.getSink(); if (sourceSpec.getInputSpecsCount() == 0) { return Function.FunctionDetails.ComponentType.SOURCE; } if (!isEmpty(sinkSpec.getBuiltin())) { return Function.FunctionDetails.ComponentType.SINK; } if (isEmpty(sinkSpec.getClassName()) || sinkSpec.getClassName().equals(PulsarSink.class.getName())) { return Function.FunctionDetails.ComponentType.FUNCTION; } return Function.FunctionDetails.ComponentType.SINK; } }
InstanceUtils { public Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails) { if (functionDetails.getComponentType() != Function.FunctionDetails.ComponentType.UNKNOWN) { return functionDetails.getComponentType(); } Function.SourceSpec sourceSpec = functionDetails.getSource(); Function.SinkSpec sinkSpec = functionDetails.getSink(); if (sourceSpec.getInputSpecsCount() == 0) { return Function.FunctionDetails.ComponentType.SOURCE; } if (!isEmpty(sinkSpec.getBuiltin())) { return Function.FunctionDetails.ComponentType.SINK; } if (isEmpty(sinkSpec.getClassName()) || sinkSpec.getClassName().equals(PulsarSink.class.getName())) { return Function.FunctionDetails.ComponentType.FUNCTION; } return Function.FunctionDetails.ComponentType.SINK; }  }
InstanceUtils { public Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails) { if (functionDetails.getComponentType() != Function.FunctionDetails.ComponentType.UNKNOWN) { return functionDetails.getComponentType(); } Function.SourceSpec sourceSpec = functionDetails.getSource(); Function.SinkSpec sinkSpec = functionDetails.getSink(); if (sourceSpec.getInputSpecsCount() == 0) { return Function.FunctionDetails.ComponentType.SOURCE; } if (!isEmpty(sinkSpec.getBuiltin())) { return Function.FunctionDetails.ComponentType.SINK; } if (isEmpty(sinkSpec.getClassName()) || sinkSpec.getClassName().equals(PulsarSink.class.getName())) { return Function.FunctionDetails.ComponentType.FUNCTION; } return Function.FunctionDetails.ComponentType.SINK; }  static SerDe<?> initializeSerDe(String serdeClassName, ClassLoader clsLoader, Class<?> typeArg,
                                           boolean deser); static Schema<?> initializeCustomSchema(String schemaClassName, ClassLoader clsLoader, Class<?> typeArg,
                                                   boolean input); Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails); static String getDefaultSubscriptionName(String tenant, String namespace, String name); static String getDefaultSubscriptionName(Function.FunctionDetails functionDetails); static Map<String, String> getProperties(Function.FunctionDetails.ComponentType componentType,
                                                    String fullyQualifiedName, int instanceId); }
InstanceUtils { public Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails) { if (functionDetails.getComponentType() != Function.FunctionDetails.ComponentType.UNKNOWN) { return functionDetails.getComponentType(); } Function.SourceSpec sourceSpec = functionDetails.getSource(); Function.SinkSpec sinkSpec = functionDetails.getSink(); if (sourceSpec.getInputSpecsCount() == 0) { return Function.FunctionDetails.ComponentType.SOURCE; } if (!isEmpty(sinkSpec.getBuiltin())) { return Function.FunctionDetails.ComponentType.SINK; } if (isEmpty(sinkSpec.getClassName()) || sinkSpec.getClassName().equals(PulsarSink.class.getName())) { return Function.FunctionDetails.ComponentType.FUNCTION; } return Function.FunctionDetails.ComponentType.SINK; }  static SerDe<?> initializeSerDe(String serdeClassName, ClassLoader clsLoader, Class<?> typeArg,
                                           boolean deser); static Schema<?> initializeCustomSchema(String schemaClassName, ClassLoader clsLoader, Class<?> typeArg,
                                                   boolean input); Function.FunctionDetails.ComponentType calculateSubjectType(Function.FunctionDetails functionDetails); static String getDefaultSubscriptionName(String tenant, String namespace, String name); static String getDefaultSubscriptionName(Function.FunctionDetails functionDetails); static Map<String, String> getProperties(Function.FunctionDetails.ComponentType componentType,
                                                    String fullyQualifiedName, int instanceId);  }