@@ -0,0 +1,3124 @@
+/**%0A * Licensed to the Apache Software Foundation (ASF) under one or more%0A * contributor license agreements.  See the NOTICE file distributed with%0A * this work for additional information regarding copyright ownership.%0A * The ASF licenses this file to You under the Apache License, Version 2.0%0A * (the %22License%22); you may not use this file except in compliance with%0A * the License.  You may obtain a copy of the License at%0A * %3Cp%3E%0A * http://www.apache.org/licenses/LICENSE-2.0%0A * %3Cp%3E%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0A%0Apackage org.apache.kafka.streams.kstream.internals;%0A%0Aimport org.apache.kafka.common.serialization.Serdes;%0Aimport org.apache.kafka.common.utils.Utils;%0Aimport org.apache.kafka.streams.KeyValue;%0Aimport org.apache.kafka.streams.kstream.KStreamBuilder;%0Aimport org.apache.kafka.streams.kstream.KeyValueMapper;%0Aimport org.apache.kafka.test.KStreamTestDriver;%0Aimport org.apache.kafka.test.MockProcessorSupplier;%0Aimport org.junit.After;%0Aimport org.junit.Before;%0Aimport org.junit.Test;%0A%0Aimport java.io.File;%0Aimport java.io.IOException;%0Aimport java.nio.file.Files;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0A%0Aimport static org.junit.Assert.assertEquals;%0A%0A%0Apublic class KGroupedTableImplTest %7B%0A%0A    private File stateDir;%0A%0A    @Before%0A    public void setUp() throws IOException %7B%0A        stateDir = Files.createTempDirectory(%22test%22).toFile();%0A    %7D%0A%0A    @After%0A    public void tearDown() throws IOException %7B%0A        Utils.delete(stateDir);%0A    %7D%0A%0A    @SuppressWarnings(%22unchecked%22)%0A    @Test%0A    public void testGroupedCountOccurences() throws IOException %7B%0A        final KStreamBuilder builder = new KStreamBuilder();%0A        final String input = %22count-test-input%22;%0A        final MockProcessorSupplier processorSupplier = new MockProcessorSupplier%3C%3E();%0A%0A        builder.table(Serdes.String(), Serdes.String(), input)%0A                .groupBy(new KeyValueMapper%3CString, String, KeyValue%3CString, String%3E%3E() %7B%0A                    @Override%0A                    public KeyValue%3CString, String%3E apply(final String key, final String value) %7B%0A                        return new KeyValue%3C%3E(value, value);%0A                    %7D%0A                %7D, Serdes.String(), Serdes.String())%0A                .count(%22count%22)%0A                .toStream()%0A                .process(processorSupplier);%0A%0A%0A        final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);%0A%0A%0A        driver.process(input, %22A%22, %22green%22);%0A        driver.process(input, %22B%22, %22green%22);%0A        driver.process(input, %22A%22, %22blue%22);%0A        driver.process(input, %22C%22, %22yellow%22);%0A        driver.process(input, %22D%22, %22green%22);%0A%0A        final List%3CString%3E expected = Arrays.asList(%22green:1%22, %22green:2%22, %22blue:1%22, %22green:1%22, %22yellow:1%22, %22green:2%22);%0A        final List%3CString%3E actual = processorSupplier.processed;%0A        assertEquals(expected, actual);%0A    %7D%0A%7D

add unit test for KGroupedTable.count