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;
        }
        // Now its between sink and function

        if (!isEmpty(sinkSpec.getBuiltin())) {
            // if its built in, its a sink
            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;
    }
@Test
    public void testCalculateSubjectTypeForSink() {
        FunctionDetails.Builder builder = FunctionDetails.newBuilder();
        // an input and a sink classname is a sink
        builder.setSource(
                Function.SourceSpec.newBuilder().putInputSpecs("topic", Function.ConsumerSpec.newBuilder().build())
                        .build());
        builder.setSink(Function.SinkSpec.newBuilder().setClassName("something").build());
        assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.SINK);
        // make sure that if the componenttype is set, that gets precedence.
        builder.setComponentType(FunctionDetails.ComponentType.SOURCE);
        assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.SOURCE);
        builder.setComponentType(FunctionDetails.ComponentType.FUNCTION);
        assertEquals(InstanceUtils.calculateSubjectType(builder.build()), FunctionDetails.ComponentType.FUNCTION);
    }