Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java (Java)

Code before:
                        public KeyValueIterator<Bytes, byte[]> all() {
                            throw new UnsupportedOperationException();
                        }

                        @Override
                        public long approximateNumEntries() {
                            return map.size();
                        }
                    };
                }

                @Override
                public String metricsScope() {
                    return "nonquery";
                }
            })
        );

        // Discard the basic streams and replace with test-specific topology
        if (kafkaStreams != null) {
            kafkaStreams.close();
        }
        final String safeTestName = safeUniqueTestName(testInfo);
        kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration(safeTestName));
        kafkaStreams.cleanUp();
        this.groupProtocol = groupProtocol;

        kafkaStreams.start();
        final StateQueryResult<ValueAndTimestamp<Integer>> result =
            IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);

        final QueryResult<ValueAndTimestamp<Integer>> queryResult =
            result.getPartitionResults().get(partition);
        assertThat(queryResult.isFailure(), is(true));
        assertThat(queryResult.getFailureReason(), is(FailureReason.UNKNOWN_QUERY_TYPE));
        assertThat(queryResult.getFailureMessage(), matchesPattern(
            "This store (.*) doesn't know how to execute the given query (.*)."
                + " Contact the store maintainer if you need support for a new query type."
        ));
    }


    private Properties streamsConfiguration(final String safeTestName) {
        final Properties config = new Properties();
        config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName + "-" + groupProtocol);
        config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:" + (++port));
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());

Review comment (bug):
The groupProtocol is being set after kafkaStreams is created and cleaned up, but the streamsConfiguration method that uses this.groupProtocol is called during kafkaStreams creation on line 448. This means the groupProtocol will be null during configuration, causing incorrect behavior.
```suggestion
        this.groupProtocol = groupProtocol;
        kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration(safeTestName));
        kafkaStreams.cleanUp();
```

Code after:
                        public KeyValueIterator<Bytes, byte[]> all() {
                            throw new UnsupportedOperationException();
                        }

                        @Override
                        public long approximateNumEntries() {
                            return map.size();
                        }
                    };
                }

                @Override
                public String metricsScope() {
                    return "nonquery";
                }
            })
        );

        // Discard the basic streams and replace with test-specific topology
        if (kafkaStreams != null) {
            kafkaStreams.close();
        }
        final String safeTestName = safeUniqueTestName(testInfo);
        this.groupProtocol = groupProtocol;
        kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration(safeTestName));
        kafkaStreams.cleanUp();

        kafkaStreams.start();
        final StateQueryResult<ValueAndTimestamp<Integer>> result =
            IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);

        final QueryResult<ValueAndTimestamp<Integer>> queryResult =
            result.getPartitionResults().get(partition);
        assertThat(queryResult.isFailure(), is(true));
        assertThat(queryResult.getFailureReason(), is(FailureReason.UNKNOWN_QUERY_TYPE));
        assertThat(queryResult.getFailureMessage(), matchesPattern(
            "This store (.*) doesn't know how to execute the given query (.*)."
                + " Contact the store maintainer if you need support for a new query type."
        ));
    }


    private Properties streamsConfiguration(final String safeTestName) {
        final Properties config = new Properties();
        config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName + "-" + groupProtocol);
        config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:" + (++port));
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());