[schema] introduce AUTO_PRODUCE schema - #2685
Conversation
*Motivation* Java instance main is expecting health check interval in seconds. *Changes Make sure the health check interval is passed in seconds. *Tests* It fixes following issue ``` 14:36:29.528 [FunctionActionerThread] INFO org.apache.pulsar.functions.runtime.ProcessRuntime - Started process successfully Exception in thread "main" java.lang.IllegalArgumentException at java.util.concurrent.ScheduledThreadPoolExecutor.scheduleAtFixedRate(ScheduledThreadPoolExecutor.java:565) at java.util.concurrent.Executors$DelegatedScheduledExecutorService.scheduleAtFixedRate(Executors.java:735) at org.apache.pulsar.functions.runtime.JavaInstanceMain.start(JavaInstanceMain.java:154) at org.apache.pulsar.functions.runtime.JavaInstanceMain.main(JavaInstanceMain.java:185) ```
*Motivation* Currently trigger function is broken due to the schema enforcement we added recently: A producer without schema can't produce messages into a topic with schema. *Changes* - Rename `AUTO` to `AUTO_CONSUME` - Introduce `AUTO_PRODUCE` schema. The schema produces `byte[]`, but it will validate the bytes are compatible with the schema associated with the topic before producing. - Change trigger function to use `AUTO_PRODUCE` schema.
sijie
added
type/bug
|
this pr is based on #2684 |
|
|
||
| Consumer<GenericRecord> consumer = pulsarClient | ||
| .newConsumer(Schema.AUTO()) | ||
| .newConsumer(Schema.AUTO_CONSUME()) |
There was a problem hiding this comment.
why are these methods, and not fields? It looks weird.
There was a problem hiding this comment.
I made them methods to be consistent with AVRO and JSON.
There was a problem hiding this comment.
AVRO and JSON take arguments though. A static method that takes no arguments is effectively a constant, and if it's not, it's doing something very suspicious in the background.
Anyhow, I wouldn't hold up the change for it, but I think it would be better as a constant.
There was a problem hiding this comment.
actually it can't be a constant. because it needs to set SchemaInfo when the client fetches schema info from brokers.
There was a problem hiding this comment.
same applied to both AUTO_CONSUME() and AUTO_PRODUCE
| ensureSchemaInitialized(); | ||
|
|
||
| // verify if the message can be decoded by the underlying schema | ||
| return schema.encode(schema.decode(message)); |
There was a problem hiding this comment.
It would be a bit more efficient only decode to check, and if it can decode, return the original byte[]
schema.decode(message);
return message;
|
run java8 tests |
|
run java8 tests |
3 participants
Motivation
Currently trigger function is broken due to the schema enforcement we added recently:
A producer without schema can't produce messages into a topic with schema.
Changes
AUTOtoAUTO_CONSUMEAUTO_PRODUCEschema. The schema producesbyte[], but it will validatethe bytes are compatible with the schema associated with the topic before producing.
AUTO_PRODUCEschema.