### Question:
RoutableProtobufKafkaSourceProvider implements SourceProvider { @Override public <T> SourceFunction<T> forSpec(IngressSpec<T> spec) { KafkaIngressSpec<T> kafkaIngressSpec = asKafkaIngressSpec(spec); return delegateProvider.forSpec(kafkaIngressSpec); }  @Override SourceFunction<T> forSpec(IngressSpec<T> spec);  }### Answer:
@Test public void exampleUsage() { JsonNode ingressDefinition = loadAsJsonFromClassResource( getClass().getClassLoader(), "routable-protobuf-kafka-ingress.yaml"); JsonIngressSpec<?> spec = new JsonIngressSpec<>( ProtobufKafkaIngressTypes.ROUTABLE_PROTOBUF_KAFKA_INGRESS_TYPE, new IngressIdentifier<>(Message.class, "foo", "bar"), ingressDefinition); RoutableProtobufKafkaSourceProvider provider = new RoutableProtobufKafkaSourceProvider(); SourceFunction<?> source = provider.forSpec(spec); assertThat(source, instanceOf(FlinkKafkaConsumer.class)); }