Join GitHub today
GitHub is home to over 50 million developers working together to host and review code, manage projects, and build software together.
Sign upAdding KafkaItemWriter single step autoconfig #697
Conversation
|
Autowire in ItemKeyMapper instead of |
a9d6716
to
cbf83d0
|
|
||
| private String topic; | ||
|
|
||
| private long pollTimeOutInSeconds = 30L; |
benas
Aug 14, 2020
Collaborator
This is not used on the writer side.
This is not used on the writer side.
cppwfs
Aug 14, 2020
Author
Contributor
EeekEEk, busted on copy paste :-)
EeekEEk, busted on copy paste :-)
| @ConfigurationProperties(prefix = "spring.batch.job.kafkaitemwriter") | ||
| public class KafkaItemWriterProperties { | ||
|
|
||
| private String name; |
benas
Aug 14, 2020
•
Collaborator
A kafkaItemWriter does not have a name property (it is not an ItemStream), so this is not needed.
A kafkaItemWriter does not have a name property (it is not an ItemStream), so this is not needed.
cppwfs
Aug 14, 2020
Author
Contributor
We need the name to disable and enable auto configuration.
We need the name to disable and enable auto configuration.
| } | ||
|
|
||
| /** | ||
| * The topic name from which the messages will be read. |
benas
Aug 14, 2020
Collaborator
On the writer's side, this should be "The topic name to which messages will be written". Same thing for the getter's javadoc.
On the writer's side, this should be "The topic name to which messages will be written". Same thing for the getter's javadoc.
cppwfs
Aug 14, 2020
Author
Contributor
EEk, busted on copy paste :-)
EEk, busted on copy paste :-)
| validateProperties(kafkaItemWriterProperties); | ||
| KafkaTemplate template = new KafkaTemplate(producerFactory); | ||
| template.setDefaultTopic(kafkaItemWriterProperties.getTopic()); | ||
| return new KafkaItemWriterBuilder<Object, Map<Object, Object>>().delete(false) |
benas
Aug 14, 2020
Collaborator
Why is delete hard-coded to false here? I think it should be configurable in KafkaItemWriterProperties.
Why is delete hard-coded to false here? I think it should be configurable in KafkaItemWriterProperties.
cppwfs
Aug 14, 2020
Author
Contributor
DOH! I have a branch I haven't pushed up with that change.
DOH! I have a branch I haven't pushed up with that change.
| return new Converter() { | ||
| @Override | ||
| public Object convert(Object source) { | ||
| return ""; |
benas
Aug 14, 2020
Collaborator
This might be problematic if the key type is not a String. I would return the source object as a key without any modification. BTW, since the itemkeyMapper is mandatory, I would make the user provide it and fail fast if it's absent. Otherwise, this default mapper can be silently used with unintended side effects (in this case, it would be writing all items with an empty String as a key).
This might be problematic if the key type is not a String. I would return the source object as a key without any modification. BTW, since the itemkeyMapper is mandatory, I would make the user provide it and fail fast if it's absent. Otherwise, this default mapper can be silently used with unintended side effects (in this case, it would be writing all items with an empty String as a key).
|
I also need to qualify the use of the Converter bean. Boot out of the box enables 3, so I need to qualify the one we want. A update is on the way. |
resolves #696
The base set of properties to setup are the following to activate the writer:
If you want to have a serializer for the value or key that is different than the boot provided default
StringSerializeryou can set the following properties to the serializer needed. In the example below I use the JsonSerializer.If a user needs another itemKeyMapper besides the default provided you can create a new bean with the name
beanItemNameMapper. i.e.