KAFKA-9088: Consolidate InternalMockProcessorContext and MockInternalProcessorContext by pierDipi · Pull Request #8157 · apache/kafka

pierDipi · 2020-02-22T10:31:55Z

These changes migrate usages of InternalMockProcessorContext to use a new version of MockInternalProcessorContext.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Demonstrate the consolidation approach via: migrating KStreamSessionWindowAggregateProcessorTest to use MockInternalProcessorContext instead of InternalMockProcessorContext

…to internal-processor-context-test

…cessorContext

- Migrated AbstractKeyValueStoreTest - Migrated CachingKeyValueStoreTest - Migrated InMemoryKeyValueLoggedStoreTest - Migrated InMemoryKeyValueStoreTest - Migrated InMemoryLRUCacheStoreTest - Migrated RocksDBKeyValueStoreTest

- Migrated RocksDBTimestampedStoreTest

- Migrated InMemorySessionStoreTest - Migrated RocksDBSessionStoreTest

…sorContext

…ernalProcessorContext

…orContext

pierDipi · 2020-02-22T10:34:23Z

Call for review: @cadonna @vvcephei @guozhangwang

guozhangwang · 2020-02-23T00:11:23Z

test this please

cadonna · 2020-03-16T10:16:45Z

@pierDipi Sorry for the late reply. I will try to review this PR this week.

…or-context-test

cadonna

@pierDipi Thank you for the PR and Sorry again for the late review.

I did a first pass and here is my feedback.

cadonna · 2020-03-05T10:14:05Z

+    }
+
+    private void init(final Properties config) {
+        init(config, createMetrics());


req: Could you try to re-use the metrics that are created in the MockProcessorContext instead of creating an additional instance here?
You probably also need to modify StreamsTestUtils#containsMetric() to use the map returned from StreamsMetrics#metrics() where the StreamsMetrics instance is returned from InternalProcessorContext#metrics().

cadonna · 2020-03-05T13:58:08Z

+        init(config);
+    }
+
+    public MockInternalProcessorContext(final StreamsMetricsImpl metrics, final ThreadCache threadCache) {


req: I looked a bit into the usages of this constructor and I think we can get rid of it. If we can get rid of it, we could also simply reuse the metrics created in MockProcessorContext instead of creating here another instance.

cadonna · 2020-03-20T10:34:36Z

    public MockInternalProcessorContext() {
+        super(StreamsTestUtils.getStreamsConfig(), DEFAULT_TASK_ID, TestUtils.tempDirectory());
+        init(StreamsTestUtils.getStreamsConfig());
+    }
+
+    public MockInternalProcessorContext(final Properties config) {
+        this(config, createStateDir(config));
+    }
+
+    public MockInternalProcessorContext(final File stateDir) {
+        this(StreamsTestUtils.getStreamsConfig(), stateDir);
+    }
+
+    public MockInternalProcessorContext(final Properties config, final File stateDir) {
+        super(config, DEFAULT_TASK_ID, stateDir);
+        init(config);
+    }
+
+    public MockInternalProcessorContext(final StreamsMetricsImpl metrics, final ThreadCache threadCache) {
+        super(StreamsTestUtils.getStreamsConfig(), DEFAULT_TASK_ID, TestUtils.tempDirectory());
+        init(metrics, threadCache);
+    }
+
+    public MockInternalProcessorContext(final LogContext logContext, final long maxCacheSizeBytes) {
+        super(StreamsTestUtils.getStreamsConfig(), DEFAULT_TASK_ID, TestUtils.tempDirectory());
+        final Metrics metrics = createMetrics();
+        final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, DEFAULT_CLIENT_ID, DEFAULT_METRICS_VERSION);
+        final ThreadCache threadCache = createThreadCache(StreamsTestUtils.getStreamsConfig(), logContext, maxCacheSizeBytes, metrics);
+        init(streamsMetrics, threadCache);
+    }
+
+    public MockInternalProcessorContext(final Properties config, final TaskId taskId, final Metrics metrics) {
+        super(config, taskId, createStateDir(config));
+        init(config, metrics);
+    }


req: Could try to reduce the number of constructors in general? I guess their number grew with the time and maybe you are able to eliminate some of them.

cadonna · 2020-03-20T10:39:41Z

+                recordContext.topic(),
+                recordContext.partition(),
+                recordContext.offset(),
+                recordContext.headers(),
+                recordContext.timestamp()


req: The indentation was correct here. Please remove the superfluous spaces.

cadonna · 2020-03-20T13:24:03Z

+    private List<KeyValueTimestamp<Object, Object>> results() {
+        return context.forwarded().stream().map(cf -> new KeyValueTimestamp<>(cf.keyValue().key, cf.keyValue().value, cf.timestamp())).collect(Collectors.toList());
+    }


prop: Here you could fix the generics, but that would mean a bit of work in MockProcessorContext, I guess. This is not required for this PR, but if you want you are very welcome to fix this in new PR. I opened ticket KAFKA-9738 to document this issue.

cadonna · 2020-03-20T13:42:58Z

+    private boolean containsMetric(final StreamsMetricsImpl streamsMetrics,
+                                   final String name,
+                                   final Map<String, String> tags) {
+        final MetricName metricName = new MetricName(name, METRIC_GROUP, "", tags);
+        return streamsMetrics.metrics().containsKey(metricName);
+    }


req: Add this to StreamsTestUtils, please. This is the method you need that I mentioned in a comment on MockInternalProcessorContext.

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

pierDipi · 2020-04-26T20:59:04Z

-        );
+        final Properties props = StreamsTestUtils.getStreamsConfig();
+        props.put(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, builtInMetricsVersion);
+        context = new MockInternalProcessorContext(props, metrics);


I didn't find a way of re-using the MockProcessorContext's StreamsMetricsImpl instance because of this test that needs to call Metrics#addReporter().
If we find a way we can get rid of the constructor MockInternalProcessorContext(Properties, Metrics)

comment: #8157 (comment)

Thank you for pointing this out. I will look into it.

pierDipi · 2020-04-26T21:00:44Z

Hi @cadonna, the PR is ready for another pass.

cadonna · 2020-04-30T10:15:52Z

@pierDipi Thank you for the updates. I will try to review it next week.

cadonna

Thank you for the updates @pierDipi !

Sorry that I haven't reviewed your PR before but I simply did not find the time.

I think the PR evolves nicely.

I saw a lot of wrong indentations throughout the PR. Could you please fix those?

You also need to rebase. Be aware that we needed to revert a change regarding the ProcessorContext that might affect this PR. So you might get some compilation error after rebasing.

Looking forward to your updates!

cadonna · 2020-05-15T11:04:46Z

+    public static final long DEFAULT_OFFSET = 0L;
+    public static final long DEFAULT_TIMESTAMP = 0L;
+    public static final String DEFAULT_CLIENT_ID = "client-id";
+    public static final String DEFAULT_THREAD_CACHE_PREFIX = "testCache ";


Suggested change

public static final String DEFAULT_THREAD_CACHE_PREFIX = "testCache ";

public static final String DEFAULT_THREAD_CACHE_PREFIX = "testCache";

cadonna · 2020-05-15T11:46:14Z

+    public static final long DEFAULT_OFFSET = 0L;
+    public static final long DEFAULT_TIMESTAMP = 0L;


prop: Could you set these two default values to something not equal 0, e.g., 42? Just to avoid the ambiguity when something in a test is initialized to 0. Also the two values should be distinct. In that way, we would immediately discover if we compare offsets with timestamps by mistake.

cadonna · 2020-05-15T11:47:19Z

+
+    public static final TaskId DEFAULT_TASK_ID = new TaskId(0, 0);
+    public static final RecordHeaders DEFAULT_HEADERS = new RecordHeaders();
+    public static final String DEFAULT_TOPIC = "";


prop: Could you give the default topic a non-empty string like "test-topic"?

cadonna · 2020-05-15T11:55:06Z

+public class MockInternalProcessorContext extends MockProcessorContext implements InternalProcessorContext<Object, Object> {
+
+    public static final TaskId DEFAULT_TASK_ID = new TaskId(0, 0);
+    public static final RecordHeaders DEFAULT_HEADERS = new RecordHeaders();


prop:

Suggested change

public static final RecordHeaders DEFAULT_HEADERS = new RecordHeaders();

public static final RecordHeaders DEFAULT_HEADERS = new RecordHeaders(

Collections.singletonList(new RecordHeader("headerKey", "headerValue".getBytes()))

);

cadonna · 2020-05-15T12:13:21Z

+        verifyDefaultMetricsVersion(context);
+        verifyDefaultRecordCollector(context);
+        verifyDefaultTaskId(context);
+        verifyDefaultTopic(context);
+        verifyDefaultPartition(context);
+        verifyDefaultTimestamp(context);
+        verifyDefaultOffset(context);
+        verifyDefaultHeaders(context);
+        verifyDefaultProcessorNodeName(context);


req: Please do not add a method for each default value. IMO using assertThat() as I did in the following is readable enough.

Suggested change

verifyDefaultMetricsVersion(context);

verifyDefaultRecordCollector(context);

verifyDefaultTaskId(context);

verifyDefaultTopic(context);

verifyDefaultPartition(context);

verifyDefaultTimestamp(context);

verifyDefaultOffset(context);

verifyDefaultHeaders(context);

verifyDefaultProcessorNodeName(context);

assertThat(context.metrics().version(), is(Version.LATEST));

assertThat(context.recordCollector(), notNullValue());

assertThat(context.taskId(), is(MockInternalProcessorContext.DEFAULT_TASK_ID));

assertThat(context.recordContext().topic(), is(MockInternalProcessorContext.DEFAULT_TOPIC));

assertThat(context.recordContext().partition(), is(MockInternalProcessorContext.DEFAULT_PARTITION));

assertThat(context.recordContext().timestamp(), is(MockInternalProcessorContext.DEFAULT_TIMESTAMP));

assertThat(context.recordContext().offset(), is(MockInternalProcessorContext.DEFAULT_OFFSET));

assertThat(context.recordContext().headers(), is(MockInternalProcessorContext.DEFAULT_HEADERS));

assertThat(context.currentNode().name(), is(MockInternalProcessorContext.DEFAULT_PROCESSOR_NODE_NAME));

cadonna · 2020-05-15T13:42:35Z

        final StoreBuilder<WindowStore<String, String>> storeBuilder = Stores.windowStoreBuilder(
-            Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false),
-            Serdes.String(),
-            Serdes.String())
-            .withCachingEnabled();
+                Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false),
+                Serdes.String(),
+                Serdes.String())
+                .withCachingEnabled();


prop: Should be

final StoreBuilder<WindowStore<String, String>> storeBuilder = Stores.windowStoreBuilder( Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false), Serdes.String(), Serdes.String() ).withCachingEnabled();

cadonna · 2020-05-15T13:44:49Z

-            Consumed.with(Serdes.String(), Serdes.String()))
-            .transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
-                private WindowStore<String, String> store;
-                private int numRecordsProcessed;
-                private ProcessorContext context;
-
-                @SuppressWarnings("unchecked")
-                @Override
-                public void init(final ProcessorContext processorContext) {
-                    this.context = processorContext;
-                    this.store = (WindowStore<String, String>) processorContext.getStateStore("store-name");
-                    int count = 0;
-
-                    final KeyValueIterator<Windowed<String>, String> all = store.all();
-                    while (all.hasNext()) {
-                        count++;
-                        all.next();
+                Consumed.with(Serdes.String(), Serdes.String()))
+                .transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
+                    private WindowStore<String, String> store;
+                    private int numRecordsProcessed;
+                    private ProcessorContext context;
+
+                    @SuppressWarnings("unchecked")
+                    @Override
+                    public void init(final ProcessorContext processorContext) {
+                        this.context = processorContext;
+                        this.store = (WindowStore<String, String>) processorContext.getStateStore("store-name");
+                        int count = 0;
+
+                        final KeyValueIterator<Windowed<String>, String> all = store.all();
+                        while (all.hasNext()) {
+                            count++;
+                            all.next();
+                        }
+
+                        assertThat(count, equalTo(0));
                    }

-                    assertThat(count, equalTo(0));
-                }
+                    @Override
+                    public KeyValue<String, String> transform(final String key, final String value) {
+                        int count = 0;

-                @Override
-                public KeyValue<String, String> transform(final String key, final String value) {
-                    int count = 0;
+                        final KeyValueIterator<Windowed<String>, String> all = store.all();
+                        while (all.hasNext()) {
+                            count++;
+                            all.next();
+                        }
+                        assertThat(count, equalTo(numRecordsProcessed));

-                    final KeyValueIterator<Windowed<String>, String> all = store.all();
-                    while (all.hasNext()) {
-                        count++;
-                        all.next();
-                    }
-                    assertThat(count, equalTo(numRecordsProcessed));
+                        store.put(value, value, context.timestamp());

-                    store.put(value, value, context.timestamp());
+                        numRecordsProcessed++;

-                    numRecordsProcessed++;
-
-                    return new KeyValue<>(key, value);
-                }
+                        return new KeyValue<>(key, value);
+                    }

-                @Override
-                public void close() {}
-            }, "store-name");
+                    @Override
+                    public void close() {
+                    }


req: It seems something happened with the indentation. I think before it was OK. In any case indentation should be 4 spaces.

cadonna · 2020-05-15T13:45:24Z

+                Serdes.String().serializer(),
+                Serdes.String().serializer(),
+                initialWallClockTime,
+                Duration.ZERO);


Suggested change

Serdes.String().serializer(),

Serdes.String().serializer(),

initialWallClockTime,

Duration.ZERO);

Serdes.String().serializer(),

Serdes.String().serializer(),

initialWallClockTime,

Duration.ZERO

);

The rest of the class seem also to have the wrong indentation.

cadonna · 2020-05-15T13:47:51Z

-        );
+        final Properties props = StreamsTestUtils.getStreamsConfig();
+        props.put(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, builtInMetricsVersion);
+        context = new MockInternalProcessorContext(props, metrics);


Thank you for pointing this out. I will look into it.

cadonna · 2020-05-15T13:50:31Z


-    public MockInternalProcessorContext(final Properties config, final TaskId taskId, final File stateDir) {
-        super(config, taskId, stateDir);
+    public MockInternalProcessorContext(final Properties config, final File stateDir, final ThreadCache cache) {


Q: Do you think we can eliminate the stateDir parameter in this constructor and just take the state dir in the config parameter?

cadonna · 2020-05-18T15:06:58Z

@pierDipi FYI: We need to make some changes to the InternalProcessorContext that are needed for two high priority tickets. The changes are done in PR #8669. The changes affect also this PR.

pierDipi · 2021-09-24T23:32:51Z

Sorry, I haven't had time to continue this contribution. :(

John Roesler and others added 30 commits January 17, 2020 12:53

POC: Consolidate testing InternalProcessorContexts

22726c6

Demonstrate the consolidation approach via: migrating KStreamSessionWindowAggregateProcessorTest to use MockInternalProcessorContext instead of InternalMockProcessorContext

Merge remote-tracking branch 'vvcephei/internal-processor-context' in…

6993a8e

…to internal-processor-context-test

Migrate WindowBytesStoreTest to use MockInternalProcessorContext

71c9936

Migrate KTableReduceTest to use MockInternalProcessorContext

45d0d38

Migrate ProcessorNodeTest to use MockInternalProcessorContext

6aa575f

Migrate SourceNodeTest to use MockInternalProcessorContext

70c9208

Migrate AbstractRocksDBSegmentedBytesStoreTest to use MockInternalPro…

d880287

…cessorContext

Migrate GlobalStateManagerImplTest to use MockInternalProcessorContext

90ccb7e

Migrate MockInternalProcessorContext to use MockInternalProcessorContex

fcc01c3

Migrate SinkNodeTest to use MockInternalProcessorContext

d1be3bd

Migrate CachingKeyValueStoreTest to use MockInternalProcessorContext

3b81c8a

Migrate CachingWindowStoreTest to use MockInternalProcessorContext

80fda2b

Migrate KeyValueSegmentsTest to use MockInternalProcessorContext

61f1eec

Fix checkstyle errors

bd6623d

Fix checkstyle errors

bdecb0f

Fix failing tests

eda30de

Refactor MockInternalProcessorContext

3126a89

Migrate MeteredWindowStoreTest to use MockInternalProcessorContext

59cdd9a

Migrate RocksDBStoreTest to use MockInternalProcessorContext

91816f6

- Migrated RocksDBTimestampedStoreTest

Migrate SessionBytesStoreTest to use MockInternalProcessorContext

7821d6e

- Migrated InMemorySessionStoreTest - Migrated RocksDBSessionStoreTest

Migrate StoreChangeLoggerTest to use MockInternalProcessorContext

1c28c43

Migrate TimestampedSegmentsTest to use MockInternalProcessorContext

0ee0243

Migrate CachingSessionStoreTest to use MockInternalProcessorContext

2dc1e28

Migrate ChangeLoggingKeyValueBytesStoreTest to use MockInternalProces…

2d255fa

…sorContext

Migrate ChangeLoggingTimestampedKeyValueBytesStoreTest to use MockInt…

bd7da9a

…ernalProcessorContext

Migrate CompositeReadOnlyKeyValueStoreTest to use MockInternalProcess…

3aec78f

…orContext

Migrate SegmentIteratorTest to use MockInternalProcessorContext

77cd3f6

Migrate WindowBytesStoreTest to use MockInternalProcessorContext

b63f80e

Migrate PartitionGroupTest to use MockInternalProcessorContext

baf6e9f

pierDipi added 7 commits February 21, 2020 22:39

Migrate RecordQueueTest to use MockInternalProcessorContext

2d3e750

Migrate CachingKeyValueStoreTest to use MockInternalProcessorContext

6a2d27b

Remove InternalMockProcessorContext

db53479

Add MockInternalProcessorContext unit tests

5edfb24

Rename method createInternalProcessorContext -> createInternalProcess…

36167d9

…orContext

Use StreamsTestUtils for getting streams configurations

decc2eb

Use @before instead of class constructor

d0c3333

vvcephei added the streams label Mar 2, 2020

Merge branch 'trunk' of github.com:apache/kafka into internal-process…

730bd39

…or-context-test

cadonna reviewed Mar 20, 2020

View reviewed changes

pierDipi added 7 commits April 22, 2020 19:57

Merge trunk

223f581

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Use key/value serdes of MockProcessorContext

ce096c7

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Remove some constructors

e09019f

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Use metrics map

d5a8a01

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Format

60406ef

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Make results final

62e854c

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Format

cb73fef

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

pierDipi commented Apr 26, 2020

View reviewed changes

cadonna reviewed May 15, 2020

View reviewed changes

pierDipi closed this Sep 24, 2021

	public static final String DEFAULT_THREAD_CACHE_PREFIX = "testCache ";
	public static final String DEFAULT_THREAD_CACHE_PREFIX = "testCache";

		public static final long DEFAULT_OFFSET = 0L;
		public static final long DEFAULT_TIMESTAMP = 0L;

Conversation

pierDipi commented Feb 22, 2020

Committer Checklist (excluded from commit message)

Uh oh!

pierDipi commented Feb 22, 2020

Uh oh!

guozhangwang commented Feb 23, 2020

Uh oh!

cadonna commented Mar 16, 2020

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pierDipi commented Apr 26, 2020

Uh oh!

cadonna commented Apr 30, 2020

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna commented May 18, 2020

Uh oh!

pierDipi commented Sep 24, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants