THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
It might also be necessary, to delete the state directory manually before starting the application. This will not result in data loss – the state will be recreated from the underlying changelog topic.0
How can I access record metadata?
Record metadata is accessible via ProcessorContext
that is provided via Processor#init()
, Transformer#init()
, and ValueTransformer#init()
. The context object is updated under the hood and contains the metadata for the currently processed record each time process()
is called.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class MyProcessor implements Processor<KEY_TYPE, VALUE_TYPE> {
private ProcessorContext context;
@Override
public void init(ProcessorContext context) {
this.context = context;
}
@Override
void process(KEY_TYPE key, VALUE_TYPE value) {
this.context.offset(); // returns the current record's offset
// you can also access #partition(), #timestamp(), and #topic()
}
// other methods omitted for brevity
} |
Producers
How should I set metadata.broker.list?
...