Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java (Java)

Code before:

    @Test
    public void shouldOverrideNonPrefixedCustomConfigsWithPrefixedConfigs() {
        props.put("custom.property.host", "host0");
        props.put(consumerPrefix("custom.property.host"), "host1");
        props.put(producerPrefix("custom.property.host"), "host2");
        props.put(adminClientPrefix("custom.property.host"), "host3");
        final StreamsConfig streamsConfig = new StreamsConfig(props);
        final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
        final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
        final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
        final Map<String, Object> adminConfigs = streamsConfig.getAdminConfigs(clientId);
        assertEquals("host1", consumerConfigs.get("custom.property.host"));
        assertEquals("host1", restoreConsumerConfigs.get("custom.property.host"));
        assertEquals("host2", producerConfigs.get("custom.property.host"));
        assertEquals("host3", adminConfigs.get("custom.property.host"));
    }

    @Test
    public void shouldOverrideAdminDefaultAdminClientEnableTelemetry() {
        final Map<String, Object> returnedProps = streamsConfig.getAdminConfigs(clientId);
        assertTrue((boolean) returnedProps.get(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG));
    }

    @Test
    public void testAutoCreateTopicsCannotBeOverriddenForStreamsConsumers() {
        // User tries to override the setting
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.mainConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");

        final StreamsConfig streamsConfig = new StreamsConfig(props);

        // Main consumer - verify override is ignored
        final Map<String, Object> mainConfigs = streamsConfig.getMainConsumerConfigs("group", "client", 0);
        assertEquals("false", mainConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Main consumer should not allow auto topic creation even with override");

        // Restore consumer - verify override is ignored
        final Map<String, Object> restoreConfigs = streamsConfig.getRestoreConsumerConfigs("client");
        assertEquals("false", restoreConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Restore consumer should not allow auto topic creation even with override");

        // Global consumer - verify override is ignored
        final Map<String, Object> globalConfigs = streamsConfig.getGlobalConsumerConfigs("client");
        assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Global consumer should not allow auto topic creation even with override");
    }

    @Test

Review comment (suggestion):
```suggestion
    public void shouldNotAllowToOverwriteAutoCreateTopicsForConsumers() {
```

Code after:

    @Test
    public void shouldOverrideNonPrefixedCustomConfigsWithPrefixedConfigs() {
        props.put("custom.property.host", "host0");
        props.put(consumerPrefix("custom.property.host"), "host1");
        props.put(producerPrefix("custom.property.host"), "host2");
        props.put(adminClientPrefix("custom.property.host"), "host3");
        final StreamsConfig streamsConfig = new StreamsConfig(props);
        final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
        final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
        final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
        final Map<String, Object> adminConfigs = streamsConfig.getAdminConfigs(clientId);
        assertEquals("host1", consumerConfigs.get("custom.property.host"));
        assertEquals("host1", restoreConsumerConfigs.get("custom.property.host"));
        assertEquals("host2", producerConfigs.get("custom.property.host"));
        assertEquals("host3", adminConfigs.get("custom.property.host"));
    }

    @Test
    public void shouldOverrideAdminDefaultAdminClientEnableTelemetry() {
        final Map<String, Object> returnedProps = streamsConfig.getAdminConfigs(clientId);
        assertTrue((boolean) returnedProps.get(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG));
    }

    @Test
    public void shouldNotAllowAutoCreateTopicsForConsumers_WithCommonConsumerPrefix() {
        // Test with generic consumer.* prefix (affects all consumer types)
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
            appender.setClassLogger(StreamsConfig.class, Level.ERROR);

            final StreamsConfig streamsConfig = new StreamsConfig(props);

            // Main consumer - verify override is ignored
            final Map<String, Object> mainConfigs = streamsConfig.getMainConsumerConfigs("group", "client", 0);
            assertEquals("false", mainConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Main consumer should not allow auto topic creation with consumer.* override");

            // Restore consumer - verify override is ignored
            final Map<String, Object> restoreConfigs = streamsConfig.getRestoreConsumerConfigs("client");
            assertEquals("false", restoreConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Restore consumer should not allow auto topic creation with consumer.* override");

            // Global consumer - verify override is ignored
            final Map<String, Object> globalConfigs = streamsConfig.getGlobalConsumerConfigs("client");
            assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Global consumer should not allow auto topic creation with consumer.* override");

            // Verify exactly 1 error is logged (consumer.* prefix is validated once in getCommonConsumerConfigs for each type of consumer)
            final List<String> errorMessages = appender.getMessages();