MINOR: unify calls to get committed offsets and metadata by mjsax · Pull Request #7463 · apache/kafka

mjsax · 2019-10-08T06:23:06Z

This is a follow up to #7304 and #6694 to unify the calls to consumer.committed() further and only call it once during startup instead of twice.

mjsax · 2019-10-08T06:23:55Z

        return stateMgr.changelogPartitions();
    }

-    Map<TopicPartition, Long> committedOffsetForPartitions(final Set<TopicPartition> partitions) {


This is moved into StandbyTask as we don't share it any longer for both types of tasks.

mjsax · 2019-10-08T06:24:21Z

    }

-    private boolean closeUnclean(final T task) {
+    private void closeUnclean(final T task) {


side cleanup: return type is never used.

mjsax · 2019-10-08T06:25:07Z

    void transitionToRunning(final T task) {
        log.debug("Transitioning {} {} to running", taskTypeName, task.id());
        running.put(task.id(), task);
-        task.initializeTaskTime();


we remove the method entirely and init task time in the constructor of StreamTask (note, that we don't need to init task time for StandbyTasks anyway)

This in now done implicitly with resume()

mjsax · 2019-10-08T06:28:04Z

+        }
+    }

+    private void initializeCommittedOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata) {


This is a mix of "old" initializeStateStores (the part to filter for source topic partitions that are use as changelogs) and "old" AbstractTaks#committedOffsetForPartitions)

mjsax · 2019-10-08T06:28:48Z

+        stateMgr.putOffsetLimits(committedOffsetsForChangelogs);
+    }
+
+    private void initializeTaskTime(final Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata) {


This is not private. Same implementation as before, however, we get offsetsAndMetadata passed in, and don't use the consumer to get them from the brokers.

mjsax · 2019-10-08T06:29:34Z

+
+
+    @Override
+    public boolean initializeStateStores() {


Just simplified -- moved the offset initialization into the constructor.

mjsax · 2019-10-08T06:29:57Z

-        final LinkedHashMap<MetricName, Metric> result = new LinkedHashMap<>();
-        result.putAll(adminClientMetrics);
-        return result;
+        return new LinkedHashMap<>(adminClientMetrics);


Side cleanup

Why do we need a LinkedHashMap here? Particularly because the interface doesn't suggest it needs anything more than a Map implementation? Actually I'm also wondering why we need to copy at all since we get back a read only view of the map from Admin and it appears the only consumer of adminClientMetrics immediately copies the results anyway.

Not sure about the LinkedHashMap -- maybe @guozhangwang knows more?

For why to copy: it's not a public contract that we get a read-only map and hence we need to guard the map from writes via a copy. Not sure if we could rely on an AdminClient implementation detail... \cc @cmccabe ?

Yes, the only use case I found above this code is to immediately copy it into yet another map (which is why the LinkedHashMap also doesn't make sense), so it should not matter, unless we're expecting to do something else with it down the road. Anyway, per our discussion I'm OK with this going in without fixing it, but it is low hanging fruit to reduce unnecessary work.

mjsax · 2019-10-08T06:31:08Z

Call for review @guozhangwang @cadonna @cpettitt-confluent

cadonna

@mjsax Thanks for the PR

Here my comments

cadonna · 2019-10-08T11:01:08Z

-    public void initializeTaskTime() {
-        //no-op
-    }
+    public void initializeTopology() {}


Could we also move this only to the StreamTask? Doesn't have to be in this PR.

Not really, because we call initializeTopology() within AssignedTasks#transitToRunning(T task) that does not know the type (it generic T extends Tasks) -- hence, we would need to do an instanceof call within that method what is undesired.

Oh, I see. I did not think about that dependency. Nevertheless I think somehow it would be possible to move initializeTopology() to StreamsTask without an instanceof but that needs a bit more thoughts and refactoring and it is out-of-scope for this PR.

cadonna · 2019-10-08T12:23:37Z

+            initializeCommittedOffsets(offsetsAndMetadata);
+            initializeTaskTime(offsetsAndMetadata);


Apparently it is not completely save to call instance methods from the constructor since the object and its member variables are not completely initialised yet. Static methods would be fine, though.

I do not think that initializeCommittedOffsets() is currently problematic since it only accesses member variables from AbstractTask which should be already initialised. However, initializeTaskTime() accesses partitionGroup which is created in the constructor. Since we cannot know what reorderings a compiler does, it would be better to specify initializeTaskTime() as static and pass in partitionGroup. To make initializeCommittedOffsets() future-proof, it would also be good to also transform it to a static method.

cpettitt-confluent

LGTM!

guozhangwang

Made a pass on the PR, lgtm modulo @cadonna 's comments.

guozhangwang · 2019-10-08T22:32:52Z

     */
    AbstractTask(final TaskId id,
-                 final Collection<TopicPartition> partitions,
+                 final Set<TopicPartition> partitions,


Why reducing the type generality here?

In StreamTask constructor we call consumer.committed(...) that takes a Set<TopicPartitions> -- hence, it's easier to pass in a Set from the beginning on. Second, thinking about it, a Collection is actually too generic anyway -- we would not want to assign a List of partitions that may contain duplicates IMHO.

mjsax · 2019-10-10T01:05:21Z

            try {
-                if (!entry.getValue().initializeStateStores()) {
+                final T task = entry.getValue();
+                task.initializeMetadata();


This is new -- we need to initialize the task after we created it.

mjsax · 2019-10-10T01:05:41Z

    }

+    @Override
+    public void initializeMetadata() {}


mjsax · 2019-10-10T01:06:45Z

                throw new ProcessorStateException(format("%sError while deleting the checkpoint file", logPrefix), e);
            }
        }
+        initializeMetadata();


This is also new -- we implicitly re-initialize the task on resume() -- was done "outside" before

Do we really need to re-initialize committed offsets and task time during Resume?

Yes, because in StreamTask#suspend() we call closeTopology() that calls partitionGroup.clear() that resets all times to UNKNOWN.

mjsax · 2019-10-10T01:07:28Z

-    static ProcessorTopology withRepartitionTopics(final List<ProcessorNode> processorNodes,
-                                                   final Map<String, SourceNode> sourcesByTopic,
-                                                   final Set<String> repartitionTopics) {
+    private static ProcessorTopology withRepartitionTopics(final List<ProcessorNode> processorNodes,


side cleanup

mjsax · 2019-10-10T01:07:33Z

        final ProcessorTopology topology = withSources(
            asList(source1, source2, processorStreamTime, processorSystemTime),
-            mkMap(mkEntry(topic1, (SourceNode) source1), mkEntry(topic2, (SourceNode) source2))
+            mkMap(mkEntry(topic1, source1), mkEntry(topic2, source2))


side cleanup

mjsax · 2019-10-10T01:07:37Z

        assertThat(expectedError, is(singletonList("ERROR")));
    }

-    @SuppressWarnings("unchecked")


side cleanup

mjsax · 2019-10-10T01:07:49Z

-                    throw new KafkaException("KABOOM!");
-                }
+            task.punctuate(processorStreamTime, 1, PunctuationType.STREAM_TIME, timestamp -> {
+                throw new KafkaException("KABOOM!");


side cleanup

mjsax · 2019-10-10T01:08:02Z

    }

+    @Test
+    public void shouldInitTaskTimeOnResumeWithEosDisabled() {


mjsax · 2019-10-10T01:08:44Z

    public void shouldThrowProcessorStateExceptionOnInitializeOffsetsWhenAuthorizationException() {
        final Consumer<byte[], byte[]> consumer = mockConsumerWithCommittedException(new AuthorizationException("message"));
        final StreamTask task = createOptimizedStatefulTask(createConfig(false), consumer);
+        task.initializeMetadata();


we moved initializing the offsets from initializeStateStores into initializeMetadata()

mjsax · 2019-10-10T01:08:59Z


        final ProcessorTopology topology = ProcessorTopologyFactories.with(
-            asList(source1),
+            singletonList(source1),


side cleanup

mjsax · 2019-10-10T01:13:28Z

Updated this PR. Call for review.

As an afterthought, I realized that we should not init the offsets and task time within the constructor of StreamTask because we should not have a blocking call (consumer.committed()) during rebalance phase (note, that tasks are created in onPartitionAssigned() callback).

Hence, I refactored the code and added a new unified init method to get the offsets. This also addressed @cadonna concerns about the initialization order within the constructor.

guozhangwang · 2019-10-11T23:09:37Z

@cadonna @cpettitt-confluent could you guys take another look?

cpettitt-confluent

One comment above on one of the side cleanups where I think we can go a bit further. Otherwise LGTM.

mjsax · 2019-10-14T22:00:01Z

Java 11 / 2.12 passed. Others failed -- test result not available any longer.

Retest this please.

mjsax · 2019-10-15T04:21:10Z

Java 11 / 2.13: kafka.log.LogCleanerIntegrationTest.testIsThreadFailed
Java 11 / 2.12: org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest.shouldWorkWithRebalance
Java 8 passed.

Reviewers: Chris Pettitt <cpettitt@confluent.io>, Bruno Cadonna <bruno@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax · 2019-10-15T04:38:43Z

Merged to trunk and cherry-picked to 2.4.

MINOR: unify calls to get committed offsets and metadata

63e6db6

mjsax added the streams label Oct 8, 2019

mjsax commented Oct 8, 2019

View reviewed changes

cadonna reviewed Oct 8, 2019

View reviewed changes

cpettitt-confluent approved these changes Oct 8, 2019

View reviewed changes

guozhangwang reviewed Oct 8, 2019

View reviewed changes

Github comments

c47dfc7

mjsax commented Oct 10, 2019

View reviewed changes

cpettitt-confluent approved these changes Oct 14, 2019

View reviewed changes

mjsax merged commit c55277c into apache:trunk Oct 15, 2019

mjsax added a commit that referenced this pull request Oct 15, 2019

MINOR: unify calls to get committed offsets and metadata (#7463)

89548dd

Reviewers: Chris Pettitt <cpettitt@confluent.io>, Bruno Cadonna <bruno@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax deleted the minor-improve-consumer-committed branch October 15, 2019 04:38

		initializeCommittedOffsets(offsetsAndMetadata);
		initializeTaskTime(offsetsAndMetadata);

Conversation

mjsax commented Oct 8, 2019

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!

mjsax Oct 8, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

mjsax commented Oct 8, 2019

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!

cadonna Oct 9, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cpettitt-confluent left a comment

Choose a reason for hiding this comment

Uh oh!

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

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!

mjsax commented Oct 10, 2019

Uh oh!

guozhangwang commented Oct 11, 2019

Uh oh!

mjsax Oct 8, 2019 •

edited

Loading

cadonna Oct 9, 2019 •

edited

Loading