Sync with numaflow built-in kafka source

Since the creation of this udsource, there have been multiple changes on the platform side, especially some kafka source bug fixes. We need to update here accordingly, otherwise this is not a stable version.