wire kafka messages internally