test: do not mock config store in StandaloneExecutor integration test

[vcrfxia]

Description

As described in #4114, the ksqlDB server from the latest release fails to start in headless mode. This bug eluded our integration tests for headless mode because the integration test currently mocks the KafkaConfigStore which updates the KSQL config passed to the StandaloneExecutor (the mock does not perform these updates). This PR switches the integration test to use an actual KafkaConfigStore instance, rather than a mock, to avoid similar bugs in the future.

Testing done

Test-only change.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[agavra]

LGTM!

[apurvam]

thanks for the patch @vcrfxia . Can we backport this to 5.4 too?

[vcrfxia]

Retargeted the patch at 5.4.x and will merge once the build passes.

[stevenpyzhang]

@vcrfxia since we're using an actual KafkaConfigStore, overrideBreakingConfigsWithOriginalValues is being called which is setting ksql.sink.paritions=4 which is failing the tests. This isn't a problem on master since the config has been removed, whereas the defaultLegacyValue is still 4 on 5.4

Probably just need to specify that config to 1 and it'll pass the tests.

[big-andy-coates]

thanks for the patch @vcrfxia . Can we backport this to 5.4 too?

Why backport? It's a test only change. It's not as though someone is going to change how the config topic code works in 5.4....

[big-andy-coates]

Is it worth removing any other mocks in this integration test while we're at it? They may open us to the same hole in testing in the future...

[vcrfxia]

Is it worth removing any other mocks in this integration test while we're at it?

The only other mock in this integration test is the version checker. Not sure what we'd replace that mock with, and it's also mocked in our other integration tests.

[vcrfxia]

Why backport? It's a test only change.

Because backporting the test fix has caught a bug! Turns out headless mode is broken on 5.4 for the reason the Jenkins build failed:

• StandaloneExecutor injects legacy values for the compatibility breaking configs. This legacy value is currently non-null for the sink partitions and replicas configs:, e.g.,

  ksql/ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

  Line 259 in 331e304

  4,

• These config values get picked up by the TopicCreateInjector via TopicProperties:

  ksql/ksql-engine/src/main/java/io/confluent/ksql/topic/TopicProperties.java

  Line 134 in 331e304

  if (config.values().containsKey(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY)) {

  ,

  ksql/ksql-engine/src/main/java/io/confluent/ksql/topic/TopicProperties.java

  Line 183 in 331e304

  .filter(Objects::nonNull)

• And are sent along to topic validation for CS/CT queries, where they fail (unless the source topic happens to have 4 partitions and 1 replica):

  ksql/ksql-engine/src/main/java/io/confluent/ksql/services/TopicValidationUtil.java

  Line 56 in 331e304

  "A Kafka topic with the name '%s' already exists, with different partition/replica "

[2019-12-16 13:57:28,996] ERROR Failed to start KSQL Server with query file: ./queries.sql (io.confluent.ksql.rest.server.StandaloneExecutor:124)
io.confluent.ksql.exception.KafkaTopicExistsException: A Kafka topic with the name 'test' already exists, with different partition/replica configuration than required. KSQL expects 4 partitions (topic has 1), and 1 replication factor (topic has 1).
	at io.confluent.ksql.services.TopicValidationUtil.validateTopicProperties(TopicValidationUtil.java:55)
	at io.confluent.ksql.services.TopicValidationUtil.validateTopicProperties(TopicValidationUtil.java:37)
	at io.confluent.ksql.services.SandboxedKafkaTopicClient.validateTopicProperties(SandboxedKafkaTopicClient.java:154)
	at io.confluent.ksql.services.SandboxedKafkaTopicClient.createTopic(SandboxedKafkaTopicClient.java:86)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at io.confluent.ksql.util.LimitedProxyBuilder.lambda$buildForwader$8(LimitedProxyBuilder.java:234)
	at io.confluent.ksql.util.LimitedProxyBuilder$SandboxProxy.invoke(LimitedProxyBuilder.java:312)
	at com.sun.proxy.$Proxy30.createTopic(Unknown Source)
	at io.confluent.ksql.topic.TopicCreateInjector.createTopic(TopicCreateInjector.java:190)
	at io.confluent.ksql.topic.TopicCreateInjector.injectForCreateSource(TopicCreateInjector.java:129)
	at io.confluent.ksql.topic.TopicCreateInjector.inject(TopicCreateInjector.java:91)
	at io.confluent.ksql.topic.TopicCreateInjector.inject(TopicCreateInjector.java:75)
	at io.confluent.ksql.statement.InjectorChain.inject(InjectorChain.java:42)
	at io.confluent.ksql.rest.server.StandaloneExecutor$StatementExecutor.prepare(StandaloneExecutor.java:312)
	at io.confluent.ksql.rest.server.StandaloneExecutor$StatementExecutor.execute(StandaloneExecutor.java:288)
	at io.confluent.ksql.rest.server.StandaloneExecutor.executeStatements(StandaloneExecutor.java:206)
	at io.confluent.ksql.rest.server.StandaloneExecutor.validateStatements(StandaloneExecutor.java:190)
	at io.confluent.ksql.rest.server.StandaloneExecutor.processesQueryFile(StandaloneExecutor.java:168)
	at io.confluent.ksql.rest.server.StandaloneExecutor.start(StandaloneExecutor.java:118)
	at io.confluent.ksql.rest.server.KsqlServerMain.tryStartApp(KsqlServerMain.java:73)
	at io.confluent.ksql.rest.server.KsqlServerMain.main(KsqlServerMain.java:59)

[vcrfxia]

The fix is to update the legacy values for the sink number of partitions and replicas configs to null and also backport #4119. Just made both changes, so hopefully the next Jenkins build passes.

[vcrfxia]

Actually, this is a bit of an odd fix since the intention of the legacy config values being non-null was so that existing queries issued prior to 5.3 (when the sink number of partitions and replicas were deprecated) will continue to validate the way they had been. I think this change is still sound since the new behavior (when the values are null) is to use existing values for number of partitions and replicas, and topics for existing queries exist by definition. However, maybe it's safer to limit this change to only set ksql.sink.replicas and ksql.sink.partitions to null in headless mode. Would be interested to hear what other members of @confluentinc/ksql think.

[vcrfxia]

Talked to @rodesai who pointed out the fix of updating the legacy config values to null should be safe regarding compatibility for interactive mode since the sink number of partitions and replicas configs are only used by the TopicCreatorInjector before writing to the command topic and not after, so changing the legacy values for the configs won't cause imcompatibilities when reading (and replaying) from the command topic.

[apurvam]

Talked to @rodesai who pointed out the fix of updating the legacy config values to null should be safe regarding compatibility for interactive mode since the sink number of partitions and replicas configs are only used by the TopicCreatorInjector before writing to the command topic and not after, so changing the legacy values for the configs won't cause imcompatibilities when reading (and replaying) from the command topic.

If I am reading this correctly, we are saying that if a user ran this in interactive mode pre 5.3, they would have their old values in the command topic, so the new defaults won't matter to them?

[apurvam]

Did we do a manual test of non-interactive mode with different input topic configs in 5.4 as well?

[vcrfxia]

Talked to @rodesai who pointed out the fix of updating the legacy config values to null should be safe regarding compatibility for interactive mode since the sink number of partitions and replicas configs are only used by the TopicCreatorInjector before writing to the command topic and not after, so changing the legacy values for the configs won't cause imcompatibilities when reading (and replaying) from the command topic.

If I am reading this correctly, we are saying that if a user ran this in interactive mode pre 5.3, they would have their old values in the command topic, so the new defaults won't matter to them?

That's actually not what I meant, but that's also true and another good reason this change is safe. I meant that for 5.4 in interactive mode, we only validate number of partitions and replicas at the time a command is received (i.e., before writing the command to the command topic) and not when replaying commands from the command topic as part of restarting the server, which means the default legacy values for the ksql.sink.partitions and ksql.sink.replicas configs (the ones that this PR changes from 4 and 1 to null) are not considered when replaying the command topic, and therefore this change can't affect how existing commands are replayed.

[vcrfxia]

Did we do a manual test of non-interactive mode with different input topic configs in 5.4 as well?

Yup, assuming this means verifying that the KSQL server starts in non-interactive mode on 5.4 with this fix, if topics have properties other than 4 partitions and 1 replica.

I also manually verified that an upgrade from KSQL 5.2 to 5.4 in interactive mode works as expected, with topics that have properties other than 4 partitions and 1 replica.