modify for kafka