KafkaIO.read.withDynamicRead() doesn't pick up new TopicPartitions

KafkaIO.Read().withDynamicRead() is correctly pulling messages from existing topics, but isn't picking up new topics or partitions when they are created.

Currently, this appears to be an interaction between the timer configuration, and the duration of the operating window, that is causing the problem