Validation doesn't take topic properties into account

[dankor]

Describe the bug
When I run a CSAS query it doesn't validate properties against the target topic. You can see the retention validation switch here:

final Long retentionMs = Stream.of(
	fromWithClause.retentionMs,
	fromOverrides.retentionMs,
	fromKsqlConfig.retentionMs)
	.filter(Objects::nonNull)
	.findFirst()
	.orElseGet(() -> fromSource.get().retentionMs);

To Reproduce

1. Create two topics with different retention.
2. Create a source stream based on the source topic.
3. Run "create stream ... as select ..." statement against the source stream and the target topic without specifying retention in both with clause and KSQL configuration.

Expected behavior
It will create a stream with no complain.

Actual behaviour
It will throw the similar message:

A Kafka topic with the name 'target_topic' already exists, with different partition/replica/retention configuration than required. KSQL expects 9 partitions (topic has 9), 3 replication factor (topic has 3), and 604800000 retention (topic has 2419200000)

Workarounds

• Set the target topic retention in the with clause explicitly.
• Split the statement into two separates ones: create stream + insert into.