Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java (Java)

Code before:
            
            consumer.assign(List.of(TP));
            consumer.seek(TP, 0);

            // consume and verify that values are modified by interceptors
            var records = consumeRecords(consumer, numRecords);
            for (var i = 0; i < numRecords; i++) {
                ConsumerRecord<String, String> record = records.get(i);
                assertEquals("key " + i, record.key());
                assertEquals(("value " + i + appendStr).toUpperCase(Locale.ROOT), record.value());
            }

            // commit sync and verify onCommit is called
            var commitCountBefore = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue();
            consumer.commitSync(Map.of(TP, new OffsetAndMetadata(2L, "metadata")));
            OffsetAndMetadata metadata = consumer.committed(Set.of(TP)).get(TP);
            assertEquals(2, metadata.offset());
            assertEquals("metadata", metadata.metadata());
            assertEquals(commitCountBefore + 1, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue());

            // commit async and verify onCommit is called
            var offsetsToCommit = Map.of(TP, new OffsetAndMetadata(5L, "metadata"));
            sendAndAwaitAsyncCommit(consumer, Optional.of(offsetsToCommit));
            metadata = consumer.committed(Set.of(TP)).get(TP);
            assertEquals(5, metadata.offset());
            assertEquals("metadata", metadata.metadata());
            assertEquals(commitCountBefore + 2, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue());
        }
        // cleanup
        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();
    }

    @ClusterTest
    public void testClassicConsumerInterceptorsWithWrongKeyValue() throws Exception {
        testInterceptorsWithWrongKeyValue(Map.of(
            GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT)
        ));
    }

    @ClusterTest
    public void testAsyncConsumerInterceptorsWithWrongKeyValue() throws Exception {
        testInterceptorsWithWrongKeyValue(Map.of(
            GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT)
        ));
    }

    private void testInterceptorsWithWrongKeyValue(Map<String, Object> consumerConfig) throws Exception {
        var appendStr = "mock";
        // create producer with interceptor that has different key and value types from the producer
        Map<String, Object> producerConfig = Map.of(

Review comment (suggestion):
Please update the docs of `OffsetAndMetadata#metadata()` to remind that the method never return `null`

Code after:
            
            consumer.assign(List.of(TP));
            consumer.seek(TP, 0);

            // consume and verify that values are modified by interceptors
            var records = consumeRecords(consumer, numRecords);
            for (var i = 0; i < numRecords; i++) {
                ConsumerRecord<String, String> record = records.get(i);
                assertEquals("key " + i, record.key());
                assertEquals(("value " + i + appendStr).toUpperCase(Locale.ROOT), record.value());
            }

            // commit sync and verify onCommit is called
            var commitCountBefore = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue();
            consumer.commitSync(Map.of(TP, new OffsetAndMetadata(2L, "metadata")));
            OffsetAndMetadata metadata = consumer.committed(Set.of(TP)).get(TP);
            assertEquals(2, metadata.offset());
            assertEquals("metadata", metadata.metadata());
            assertEquals(commitCountBefore + 1, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue());

            // commit async and verify onCommit is called
            var offsetsToCommit = Map.of(TP, new OffsetAndMetadata(5L, null));
            sendAndAwaitAsyncCommit(consumer, Optional.of(offsetsToCommit));
            metadata = consumer.committed(Set.of(TP)).get(TP);
            assertEquals(5, metadata.offset());
            // null metadata will be converted to an empty string
            assertEquals("", metadata.metadata());
            assertEquals(commitCountBefore + 2, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue());
        }
        // cleanup
        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();
    }

    @ClusterTest
    public void testClassicConsumerInterceptorsWithWrongKeyValue() throws Exception {
        testInterceptorsWithWrongKeyValue(Map.of(
            GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT)
        ));
    }

    @ClusterTest
    public void testAsyncConsumerInterceptorsWithWrongKeyValue() throws Exception {
        testInterceptorsWithWrongKeyValue(Map.of(
            GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT)
        ));
    }

    private void testInterceptorsWithWrongKeyValue(Map<String, Object> consumerConfig) throws Exception {
        var appendStr = "mock";
        // create producer with interceptor that has different key and value types from the producer