MINOR: consolidate processor context for active/standby by ableegoldman · Pull Request #8669 · apache/kafka

ableegoldman · 2020-05-14T22:18:52Z

This is a prerequisite for KAFKA-9501 and will also be useful for KAFKA-9603

There should be no logical changes here: the main difference is the removal of StandbyContextImpl in preparation for contexts to transition between active and standby.

Also includes some minor cleanup, eg pulling the ReadOnly/ReadWrite decorators out into a separate file.

ableegoldman · 2020-05-14T22:19:14Z

Call for review @cadonna @guozhangwang @vvcephei

ableegoldman · 2020-05-14T22:31:27Z

I felt these were just cluttering up this class so I moved them to a new file

ableegoldman · 2020-05-15T03:21:10Z

+import org.apache.kafka.streams.state.WindowStoreIterator;
+import org.apache.kafka.streams.state.internals.WrappedStateStore;
+
+abstract class AbstractReadOnlyDecorator<T extends StateStore, K, V> extends WrappedStateStore<T, K, V> {


All of this (and the other new class below) was just copied over from ProcessorContextImpl. I changed the named to AbstractXXXDecorator since it's an abstract class but it's all otherwise unchanged

ableegoldman · 2020-05-15T05:39:13Z

    }

    /**
+     * @throws UnsupportedOperationException if the current task type is standby


No logical changes here, just added a check for any methods that were previously overridden in the standby context to throw UnsupportedOperation

ableegoldman · 2020-05-15T05:40:44Z

-        logChange(key, value, context.timestamp());
-    }
-
-    void logChange(final K key,


This class was the root cause of the processor context issue blocking the active <--> standby task conversion. I was taking pieces out of it bit by bit and by the end it seemed pointless to have at all

ableegoldman · 2020-05-15T05:41:34Z

    void log(final Bytes key,
             final byte[] value) {
-        changeLogger.logChange(key, value);
+        context.logChange(name(), key, value, context.timestamp());


We now just delegate to the context to figure out how/what to log

vvcephei

Thanks for this, @ableegoldman !

I had a few comments; once they're addressed, I'll be +1 to merge this.

Thanks,
-John

vvcephei · 2020-05-15T14:57:01Z

+    @Override
+    public TaskType taskType() {
+        return stateManager.taskType();
+    }


Looks like this doesn't need to be defaulted here. If the logic doesn't apply to all the implementing classes, it's better not to define it in the abstract class.

Ah, actually it does. I should remove the override from the ProcessorContextImpl instead

vvcephei · 2020-05-15T16:04:23Z

+    BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
+    ByteArraySerializer BYTE_ARRAY_VALUE_SERIALIZER = new ByteArraySerializer();
+


It doesn't seem like these need to be defined here, since they're only used outside of this interface.

They actually only have two, independent, usages, and it doesn't seem that important to de-duplicate the instances. Can we just copy them to separate constants in the classes that need them?

Yeah, this was part of a larger refactoring that I walked back. That sounds like a reasonable request 👍

vvcephei · 2020-05-15T16:19:07Z

+        EasyMock.reset(context);
+        EasyMock.expect(context.timestamp()).andStubReturn(0L);
+        context.logChange(store.name(), key1, value, 0L);
+        context.logChange(store.name(), key2, value, 0L);


We don't need expectLastCall() on these (and everywhere else)?

The expectLastCall is redundant if you're not chaining it with something else (like expectLastCall().times(2))

I verified this just to be sure by removing the invocation of context.logChange in this class and it did indeed fail

Ok, thanks!

vvcephei · 2020-05-15T16:20:59Z

    private final List<CapturedForward> capturedForwards = new LinkedList<>();
    private boolean committed = false;

+


Not sure about this change ;)

vvcephei · 2020-05-15T16:29:44Z

test this please

vvcephei · 2020-05-15T18:07:16Z

Test this please

vvcephei · 2020-05-15T18:07:44Z

Test this please

ableegoldman · 2020-05-15T21:10:09Z

All builds failed due to some mysterious

11:59:42 FATAL: command execution failed
11:59:42 java.io.EOFException
11:59:42 	at java.io.ObjectInputStream$PeekInputStream.readFully(ObjectInputStream.java:2681)
11:59:42 	at java.io.ObjectInputStream$BlockDataInputStream.readShort(ObjectInputStream.java:3156)
11:59:42 	at java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:862)
11:59:42 	at java.io.ObjectInputStream.<init>(ObjectInputStream.java:358)
11:59:42 	at hudson.remoting.ObjectInputStreamEx.<init>(ObjectInputStreamEx.java:49)
11:59:42 	at hudson.remoting.Command.readFrom(Command.java:140)
11:59:42 	at hudson.remoting.Command.readFrom(Command.java:126)
11:59:42 	at hudson.remoting.AbstractSynchronousByteArrayCommandTransport.read(AbstractSynchronousByteArrayCommandTransport.java:36)
11:59:42 	at hudson.remoting.SynchronousCommandTransport$ReaderThread.run(SynchronousCommandTransport.java:63)
11:59:42 Caused: java.io.IOException: Unexpected termination of the channel

guozhangwang · 2020-05-15T21:15:48Z

test this please

ableegoldman · 2020-05-15T22:10:05Z

All builds failed on 14:16:00 ERROR: Error cloning remote repo 'origin' 😕

guozhangwang · 2020-05-16T00:46:12Z

test this please

guozhangwang

LGTM overall. I'd like to also ping @cadonna since he's reviewing another large PR that merges MockInternal- and InternalMock- ProcessorContext which would have some conflicts with this one.

guozhangwang · 2020-05-16T01:02:43Z

+                          final Bytes key,
+                          final byte[] value,
+                          final long timestamp) {
+        recordCollector().send(


When we merge the InternalMock with MockInternal would this be okay? cc @cadonna

I assumed this would be fine since it's pretty much what happened before (ie users of the context would get the record collector and then call send) but I'd like to get this confirmed

Both, MockInternalProcessorContext and InternalMockProcessorContext currently implement RecordCollector.Supplier (i.e., recordCollector()). Of course, once rebased the consolidated mock needs to implement taskType() and logChange().
See also my comment above regarding KEY_SERIALIZER and VALUE_SERIALIZER.
@guozhangwang did you have anything specific in mind that I did not cover here?

No I do not, just wanting to make sure we do not have any major conflicts when rebasing the other.

guozhangwang · 2020-05-16T01:03:38Z

-        logChange(key, value, context.timestamp());
-    }
-
-    void logChange(final K key,


cadonna

@ableegoldman Thank you for this! I am looking forward to use this in the fix for KAFKA-9603.

Here my feedback.

cadonna · 2020-05-18T12:35:41Z

+                          final Bytes key,
+                          final byte[] value,
+                          final long timestamp) {
+        throwUnsupportedOperationExceptionIfStandby("logChange");


req: Please unit test a processor context for a standby with the unsupported methods. That is, all code paths that involve a call to throwUnsupportedOperationExceptionIfStandby().

cadonna · 2020-05-18T13:49:48Z

+import static org.apache.kafka.streams.processor.internals.ProcessorContextImpl.KEY_SERIALIZER;
+import static org.apache.kafka.streams.processor.internals.ProcessorContextImpl.VALUE_SERIALIZER;


prop: I think we should move those to InternalProcessorContext. IMO, it would be cleaner for a mock not to have a direct dependency to the class it mocks. The new consolidated mock for the internal processor context will not extend AbstractProcessorContext but only MockProcessorContext. Thus, the common ancestor will be InternalProcessorContext.
Sorry for bothering you because of those constants after that @vvcephei has already bothered you. :-)

cadonna · 2020-05-18T13:52:14Z

+                          final Bytes key,
+                          final byte[] value,
+                          final long timestamp) {
+        recordCollector().send(


Both, MockInternalProcessorContext and InternalMockProcessorContext currently implement RecordCollector.Supplier (i.e., recordCollector()). Of course, once rebased the consolidated mock needs to implement taskType() and logChange().
See also my comment above regarding KEY_SERIALIZER and VALUE_SERIALIZER.
@guozhangwang did you have anything specific in mind that I did not cover here?

guozhangwang · 2020-05-18T16:35:09Z

test this please

guozhangwang · 2020-05-18T16:35:33Z

LGTM. Will merge after green builds.

guozhangwang · 2020-05-18T17:14:17Z

test this please

guozhangwang · 2020-05-18T18:47:02Z

test this please

ableegoldman · 2020-05-18T21:37:01Z

Failed due to flaky EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta and ConnectorTopicsIntegrationTest.testGetActiveTopics

ableegoldman commented May 14, 2020

View reviewed changes

ableegoldman force-pushed the MINOR-consolidate-ProcessorContext-impls branch from 62d7079 to 751e8e8 Compare May 15, 2020 00:08

ableegoldman added 5 commits May 14, 2020 17:09

add taskType method

10b2068

removed StandbyContextImpl

cbb68fd

checkstyle

0600e5c

fix mocks

7aeaa20

return state store

2a506c4

ableegoldman force-pushed the MINOR-consolidate-ProcessorContext-impls branch from 751e8e8 to 2a506c4 Compare May 15, 2020 00:09

ableegoldman added 3 commits May 14, 2020 17:36

fix stupid spotBugs nonsense >:(

fe5182e

fix tests

3bd771b

fix context tests

a5b4cee

ableegoldman commented May 15, 2020

View reviewed changes

further refactoring

632617b

ableegoldman commented May 15, 2020

View reviewed changes

vvcephei reviewed May 15, 2020

View reviewed changes

ableegoldman added 2 commits May 15, 2020 10:56

github feedback

efc5306

checkstyle

a1e0842

guozhangwang reviewed May 16, 2020

View reviewed changes

github change

94bc716

cadonna reviewed May 18, 2020

View reviewed changes

cadonna mentioned this pull request May 18, 2020

KAFKA-9088: Consolidate InternalMockProcessorContext and MockInternalProcessorContext #8157

Closed

3 tasks

ableegoldman added 2 commits May 18, 2020 09:00

add tests and move constants

ea029f2

checkstyle

67c3e05

fix standby test

f807b4f

guozhangwang merged commit 392e49b into apache:trunk May 18, 2020

ableegoldman deleted the MINOR-consolidate-ProcessorContext-impls branch May 18, 2020 23:30

		BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
		ByteArraySerializer BYTE_ARRAY_VALUE_SERIALIZER = new ByteArraySerializer();

		private final List<CapturedForward> capturedForwards = new LinkedList<>();
		private boolean committed = false;

		import static org.apache.kafka.streams.processor.internals.ProcessorContextImpl.KEY_SERIALIZER;
		import static org.apache.kafka.streams.processor.internals.ProcessorContextImpl.VALUE_SERIALIZER;

Conversation

ableegoldman commented May 14, 2020

Uh oh!

ableegoldman commented May 14, 2020

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!

vvcephei 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!

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!

Choose a reason for hiding this comment

Uh oh!

vvcephei commented May 15, 2020

Uh oh!

vvcephei commented May 15, 2020

Uh oh!

vvcephei commented May 15, 2020

Uh oh!

ableegoldman commented May 15, 2020

Uh oh!

guozhangwang commented May 15, 2020

Uh oh!

ableegoldman commented May 15, 2020

Uh oh!

guozhangwang commented May 16, 2020

Uh oh!

guozhangwang left a comment

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

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!

guozhangwang commented May 18, 2020

Uh oh!

guozhangwang commented May 18, 2020