The Metrics ReplicationBytesInPerSec does not have label of topic

Hi, I found that the metcis kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec only exposes on allTopicsStates, so that it cannot expose the metrics in Topic level. brokerTopicStats.updateReplicationBytesIn(records.sizeInBytes) def updateReplicationBytesIn(value: Long): Unit = { allTop

Stream sinks are not constructed when application starts up before Kafka broker

My team is building a set of services that use Kafka Connect and Debezium to forward data changes from our Postgres database to Kafka, and then use Kafka Streams (via Spring Cloud Stream) to process this data and output an aggregate of the source data. We have been trying to track down an issue wh