KAFKA-12849: KIP-744 TaskMetadata ThreadMetadata StreamsMetadata as API #10840

jlprat · 2021-06-08T12:26:43Z

Implementation of KIP-744.

Creates new Interfaces for TaskMetadata, ThreadMetadata, and
StreamsMetadata, providing internal implementations for each of them.

Deprecates current TaskMetadata, ThreadMetadata under o.a.k.s.processor,
and SreamsMetadata under a.o.k.s.state.

Updates references on internal classes from deprecated classes to new interfaces.

Deprecates methods on KStreams returning deprecated ThreadMeatada and
StreamsMetadta, and provides new ones returning the new interfaces.

Update Javadocs referencing to deprecated classes and methods to point
to the right ones.

Committer Checklist (excluded from commit message)

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

jlprat

Added some notes to help navigate the PR

jlprat · 2021-06-08T12:32:38Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     *
+     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances of this application
+     */
+    public Collection<StreamsMetadata> allRunningMetadata() {


Not so much convinced about this name, open to suggestions if you don't like it.

How about allStreamsMetadata? I think that gets at the core difference here, which is that this returns metadata for all Streams instances.

What about metadataForAllStreamsClients()? I think this makes it more explicit.

I decided to for for the pattern xxxForxxx to keep consistency among different changes.

It is now metadataForAllStreamsClients but happy to change if anyone has reasons against it.

jlprat · 2021-06-08T12:33:08Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

    /**
     * Returns runtime information about the local threads of this {@link KafkaStreams} instance.
     *
     * @return the set of {@link ThreadMetadata}.
     */
-    public Set<ThreadMetadata> localThreadsMetadata() {
+    public Set<ThreadMetadata> threadsMetadata() {


Similarly here, name is maybe not the best, happy to change it if there is a better alternative.

I think localThreadMetadata sounds better to me (actually even better than the original localThreadsMetadata tbh), but I'm fine with threadsMetadata too

nit: I think localThreadMetadata does not sound consistent since metadata of multiple stream threads is returned. What about metadataForLocalThreads()?

I decided to for for the pattern xxxForxxx to keep consistency among different changes.

It is now metadataForLocalThreads but happy to change if anyone has reasons against it.

@cadonna localThreadMetadata still sounds more correct to me than localThreadsMetadata. I really can't explain it other than to say that English is weird, and names/titles like this do not always follow the regular rules of grammar/plurals 🤷‍♀️

But actually I like your suggestion metadataForLocalThreads() even better than any of them, SGTM

jlprat · 2021-06-08T12:34:23Z

streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java

 public class TaskMetadata {

-    private final TaskId taskId;
+    private final String taskId;


This reverts TaskMetadata to the state before KIP-740. As this TaskMetadata is deprecated, no need to do any type change here.

jlprat · 2021-06-08T12:35:07Z

streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java

@@ -68,17 +61,8 @@ public TaskMetadata(final TaskId taskId,
    /**
     * @return the basic task metadata such as subtopology and partition id
     */
-    public TaskId getTaskId() {


Same here, this reverts to the state prior KIP-740. Users are anyway discouraged to use this class.

jlprat · 2021-06-08T12:37:45Z

streams/src/main/java/org/apache/kafka/streams/TaskMetadata.java

+    /**
+     * @return the basic task metadata such as subtopology and partition id
+     */
+    TaskId taskId();


Cleaned up API here, with proper naming convention for the new TaskMetada

jlprat · 2021-06-08T12:39:02Z

As @cadonna, @ableegoldman and @guozhangwang showed some interest on the KIP, I guess maybe any of you can review this PR?

jlprat · 2021-06-08T18:00:27Z

Failure was:

[2021-06-08T14:01:59.439Z] FAILURE: Build failed with an exception.
[2021-06-08T14:01:59.439Z] 
[2021-06-08T14:01:59.439Z] * What went wrong:
[2021-06-08T14:01:59.439Z] Execution failed for task ':core:integrationTest'.
[2021-06-08T14:01:59.439Z] > Process 'Gradle Test Executor 127' finished with non-zero exit value 1
[2021-06-08T14:01:59.439Z]   This problem might be caused by incorrect test process configuration.
[2021-06-08T14:01:59.439Z]   Please refer to the test execution section in the User Manual at https://docs.gradle.org/7.0.2/userguide/java_testing.html#sec:test_execution

On top of the common RaftClusterTest ones

jlprat · 2021-06-15T07:16:18Z

Following Konstantine's recommendation on the mailing list in regards of the dates for the release of Kafka 3.0.0, I'm pinging again potential reviewers for this KIP's implementation.
@ableegoldman (as creator of the original Jira Ticket), @cadonna and @guozhangwang (as voters of the KIP), I would highly appreciate your feedback in this PR.

Thanks a lot!

ableegoldman

Did a quick pass, looks good for the most part but I think we can take the opportunity to clean up the method names a bit. Lmk what you think of the suggestions

ableegoldman · 2021-06-23T01:23:25Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

    /**
     * Returns runtime information about the local threads of this {@link KafkaStreams} instance.
     *
     * @return the set of {@link ThreadMetadata}.
     */
-    public Set<ThreadMetadata> localThreadsMetadata() {
+    public Set<ThreadMetadata> threadsMetadata() {


I think localThreadMetadata sounds better to me (actually even better than the original localThreadsMetadata tbh), but I'm fine with threadsMetadata too

ableegoldman · 2021-06-23T01:26:33Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     *
+     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances of this application
+     */
+    public Collection<StreamsMetadata> allRunningMetadata() {


How about allStreamsMetadata? I think that gets at the core difference here, which is that this returns metadata for all Streams instances.

ableegoldman · 2021-06-23T01:29:07Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances with the provide {@code storeName} of
+     * this application
+     */
+    public Collection<StreamsMetadata> allMetadataForGivenStore(final String storeName) {


I don't feel too strongly about any of these names, but it's probably best to try and keep them (a) consistent with each other, where possible, and (b) consistent with the return type. Assuming we go with something like allStreamsMetadata for the new method above, maybe something like streamsMetadataForStore would be appropriate for this one?

I am +1 for streamsMetadataForStore.

I decided to for for the pattern xxxForxxx to keep consistency among different changes.

It is now streamsMetadataForStore but happy to change if anyone has reasons against it.

ableegoldman · 2021-06-23T01:31:02Z

streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java

@@ -27,10 +27,12 @@

 /**
 * Represents the state of a single task running within a {@link KafkaStreams} application.
+ * @deprecated since 3.0, not intended for public use, use {@link org.apache.kafka.streams.TaskMetadata} instead.


Suggested change

* @deprecated since 3.0, not intended for public use, use {@link org.apache.kafka.streams.TaskMetadata} instead.

* @deprecated since 3.0, use {@link org.apache.kafka.streams.TaskMetadata} instead.

I think people might be a bit confused if we say this, just point them to the new interface

ableegoldman · 2021-06-23T01:31:54Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java

                                                  Collections.emptySet(),
                                                  Collections.emptySet(),
                                                  Collections.emptySet(),


nit: fix alignment (here and below change)

ableegoldman · 2021-06-23T01:32:56Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskMetadataImpl.java

+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final org.apache.kafka.streams.processor.internals.TaskMetadataImpl that = (org.apache.kafka.streams.processor.internals.TaskMetadataImpl) o;


Does this need to be fully qualified? There's only one TaskMetadataImpl class right?

ableegoldman · 2021-06-23T01:33:38Z

streams/src/main/java/org/apache/kafka/streams/state/internals/StreamsMetadataImpl.java

+                                                                            Collections.emptySet(),
+                                                                            Collections.emptySet(),
+                                                                            Collections.emptySet(),
+                                                                            Collections.emptySet());
+


nit: fix alignment

streams/src/main/java/org/apache/kafka/streams/StreamsMetadata.java

cadonna

@jlprat Thank you for the PR!

Sorry for the late review!

Here my feedback.

cadonna · 2021-06-23T08:47:35Z

streams/src/main/java/org/apache/kafka/streams/TaskMetadata.java

+    /**
+     * This function will return a set of the current TopicPartitions
+     */
+    Set<TopicPartition> topicPartitions();


Could you please use javadoc mark-up like @return and @param for the docs? Here and for the other methods.

cadonna · 2021-06-23T08:52:10Z

streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java

 */
+@Deprecated


nit: If we deprecate the class, we do not need to deprecate the constructor anymore, right?

cadonna · 2021-06-23T08:55:15Z

streams/src/main/java/org/apache/kafka/streams/ThreadMetadata.java

+    /**
+     * This function will return the set of the {@link TaskMetadata} for the current active tasks
+     */
+    Set<TaskMetadata> activeTasks();


Could you use javadoc mark-up for the docs? Here and for the other methods.

streams/src/main/java/org/apache/kafka/streams/StreamsMetadata.java

cadonna · 2021-06-23T09:15:52Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

    /**
     * Returns runtime information about the local threads of this {@link KafkaStreams} instance.
     *
     * @return the set of {@link ThreadMetadata}.
     */
-    public Set<ThreadMetadata> localThreadsMetadata() {
+    public Set<ThreadMetadata> threadsMetadata() {


nit: I think localThreadMetadata does not sound consistent since metadata of multiple stream threads is returned. What about metadataForLocalThreads()?

cadonna · 2021-06-23T09:30:02Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

@@ -1972,12 +1971,6 @@ public void shouldAlwaysUpdateTasksMetadataAfterChangingState() {
        assertEquals(StreamThread.State.RUNNING.name(), metadata.threadState());
    }

-    private void assertThreadMetadataHasEmptyTasksWithState(final ThreadMetadata metadata, final StreamThread.State state) {


Thank you for the clean up!

cadonna · 2021-06-23T09:35:23Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskMetadataImpl.java

+        this.topicPartitions = topicPartitions;
+        this.committedOffsets = committedOffsets;
+        this.endOffsets = endOffsets;


Since this class is read only, it makes sense to use Collections.unmodifiable*() methods here to avoid modification of the contents of the member fields.
Unit tests as in StreamsMetadataTest would also be good.

cadonna · 2021-06-23T09:37:33Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImpl.java

+                              final Set<TaskMetadata> standbyTasks) {
+        this.mainConsumerClientId = mainConsumerClientId;
+        this.restoreConsumerClientId = restoreConsumerClientId;
+        this.producerClientIds = producerClientIds;


I think, you should use Collections.unmodifiableSet() also here.
Also unit tests to verify the immutability as in StreamsMetadataTest would be great!

cadonna · 2021-06-23T09:43:10Z

streams/src/main/java/org/apache/kafka/streams/state/internals/StreamsMetadataImpl.java

+     */
+    @Override
+    public Set<String> stateStoreNames() {
+        return Collections.unmodifiableSet(stateStoreNames);


I think it would be better to make the collections immutable in the constructor since they should also not be modified within this class (for now).

cadonna · 2021-06-23T09:45:42Z

docs/streams/upgrade-guide.html

+        <code>org.apache.kafka.streams.TaskMetadata</code> to be used instead. This change was introduced to better reflect the fact that <code>TaskMetadata</code> was not meant to be instantiated outside
+        of Kafka codebase.
+        Please note that the new <code>TaskMetadata</code> offers APIs that better represent the task id as an actual <code>TaskId</code> object instead of a String. Please migrate to the new
+        <code>org.apache.kafka.streams.TaskMetadata</code> which offers these better methods, for example, by using the new <code>ThreadMetadata#getActiveTasks</code> and <code>ThreadMetadata#getStandbyTasks</code>.


Suggested change

<code>org.apache.kafka.streams.TaskMetadata</code> which offers these better methods, for example, by using the new <code>ThreadMetadata#getActiveTasks</code> and <code>ThreadMetadata#getStandbyTasks</code>.

<code>org.apache.kafka.streams.TaskMetadata</code> which offers these better methods, for example, by using the new <code>ThreadMetadata#activeTasks</code> and <code>ThreadMetadata#standbyTasks</code>.

Implementation of KIP-744. Creates new Interfaces for TaskMetadata, ThreadMetadata, and StreamsMetadata, providing internal implementations for each of them. Deprecates current TaskMetadata, ThreadMetadata under o.a.k.s.processor, and SreamsMetadata under a.o.k.s.state. Updates references on internal classes from deprecated classes to new interfaces. Deprecates methods on KStreams returning deprecated ThreadMeatada and StreamsMetadta, and provides new ones returning the new interfaces. Update Javadocs referencing to deprecated classes and methods to point to the right ones.

jlprat · 2021-06-23T14:04:10Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImpl.java

+        if (producerClientIds != null) {
+            this.producerClientIds = Collections.unmodifiableSet(producerClientIds);
+        } else {
+            this.producerClientIds = Collections.emptySet();
+        }


@cadonna, I needed to add this guard as ThreadMetadataTest was failing with NPE.

I could not reproduce this issue. the set of producer IDs should never be null. If this is the case the test is wrong.

OK, now I see. You meant StreamThreadTest. There you need to setup the mock for the TaskManager to return at least an empty set with expect(taskManager.producerClientIds()).andStubReturn(Collections.emptySet());.

jlprat · 2021-06-23T14:06:30Z

Needed to rebase as there were some conflicts with trunk, hence the force push.

I applied the changes in separate commits:

One for method renames, formattings and small refactors (like the unmodifiable collections)
One for Adding tests
One to fix the regression caused by the unmodifiable collections in constructor.

jlprat

Added some notes about the hash code and equals testing approach

jlprat · 2021-06-23T14:07:55Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskMetadataImplTest.java

+    }
+
+    @Test
+    public void shouldFollowEqualsAndHasCodeContract() {


@cadonna For hash code and equals contract validation, I decided to keep it under one single test, and validate the positive cases and any of the reasons why it might fall into the negative case.
Please note that objects differing only on committed offsets, end offsets, and/or time curring idling started will be considered equals.

jlprat · 2021-06-23T14:09:31Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImplTest.java

+    }
+
+    @Test
+    public void shouldFollowHashCodeAndEqualsContract() {


@cadonna For hash code and equals contract validation, I decided to keep it under one single test, and validate the positive case and any of the reasons why it might fall into the negative case.

I prefer to have separate tests, because when something fails you immediately know the cause. But if you like to keep like that, I am fine with it.

jlprat · 2021-06-23T14:09:47Z

streams/src/test/java/org/apache/kafka/streams/state/StreamsMetadataTest.java

@@ -55,6 +62,63 @@ public void shouldNotAllowModificationOfInternalStateViaGetters() {
        assertTrue(isUnmodifiable(streamsMetadata.standbyStateStoreNames()));
    }

+    @Test
+    public void shouldFollowHashCodeAndEqualsContract() {


@cadonna For hash code and equals contract validation, I decided to keep it under one single test, and validate the positive case and any of the reasons why it might fall into the negative case.

see my other comment about separate tests.

jlprat · 2021-06-23T14:11:10Z

@ableegoldman and @cadonna thanks a lot for your reviews! I think I addressed all of your comments. Let me know what do you think about them.

jlprat · 2021-06-24T09:28:15Z

I see some usual suspects on test failures, but org.apache.kafka.connect.integration.SourceConnectorsIntegrationTest.testSwitchingToTopicCreationEnabledis new. Looking at the stack trace, doesn't seem relevant to the changes made in this PR.

java.lang.NullPointerException
	at java.util.concurrent.ConcurrentHashMap.get(ConcurrentHashMap.java:936)
	at org.reflections.Store.getAllIncluding(Store.java:82)
	at org.reflections.Store.getAll(Store.java:93)
	at org.reflections.Reflections.getSubTypesOf(Reflections.java:404)
	at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.getPluginDesc(DelegatingClassLoader.java:352)
	at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.scanPluginPath(DelegatingClassLoader.java:337)
	at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.scanUrlsAndAddPlugins(DelegatingClassLoader.java:268)
	at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.initPluginLoader(DelegatingClassLoader.java:216)
	at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.initLoaders(DelegatingClassLoader.java:209)
	at org.apache.kafka.connect.runtime.isolation.Plugins.<init>(Plugins.java:61)
	at org.apache.kafka.connect.cli.ConnectDistributed.startConnect(ConnectDistributed.java:93)
	at org.apache.kafka.connect.util.clusters.WorkerHandle.start(WorkerHandle.java:50)
	at org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster.addWorker(EmbeddedConnectCluster.java:174)
	at org.apache.kafka.connect.integration.SourceConnectorsIntegrationTest.lambda$testSwitchingToTopicCreationEnabled$1(SourceConnectorsIntegrationTest.java:197)
	at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)
	at java.util.stream.IntPipeline$Head.forEach(IntPipeline.java:559)
	at org.apache.kafka.connect.integration.SourceConnectorsIntegrationTest.testSwitchingToTopicCreationEnabled(SourceConnectorsIntegrationTest.java:197)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:110)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
	at org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:62)
	at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
	at sun.reflect.GeneratedMethodAccessor3.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
	at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
	at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
	at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:121)
	at sun.reflect.GeneratedMethodAccessor2.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
	at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:182)
	at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:164)
	at org.gradle.internal.remote.internal.hub.MessageHub$Handler.run(MessageHub.java:414)
	at org.gradle.internal.concurrent.ExecutorPolicy$CatchAndRecordFailures.onExecute(ExecutorPolicy.java:64)
	at org.gradle.internal.concurrent.ManagedExecutorImpl$1.run(ManagedExecutorImpl.java:48)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at org.gradle.internal.concurrent.ThreadFactoryImpl$ManagedThreadRunnable.run(ThreadFactoryImpl.java:56)
	at java.lang.Thread.run(Thread.java:748)

Is there some relation I'm missing, or should I create a new issue for this flaky test?

cadonna

@jlprat Thank you for the update!

Here my comments!

Regarding the docs of the interfaces, could you provide at least one initial sentence for each javadoc? I know that the old classes were not documented that well. If you do not have time, we can also postpone the improvements of the docs to a separate PR and get this PR merged as soon as possible.

cadonna · 2021-06-24T10:09:31Z

docs/streams/upgrade-guide.html

+        <li>Users of <code>KafkaStreams#allMetadata</code> are meant to migrate to the new <code>KafkaStreams#allRunningMetadata</code>.</li>
+        <li>Users of <code>KafkaStreams#allMetadataForStore(String)</code> are meant to migrate to the new <code>KafkaStreams#allMetadataForGivenStore(String)</code>.</li>
+        <li>Users of <code>KafkaStreams#localThreadsMetadata</code> are meant to migrate to the new <code>KafkaStreams#threadsMetadata</code>.</li>


You need to adapt this text to the new method names.

cadonna · 2021-06-24T11:24:43Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java

+            localMetadata.set(new StreamsMetadataImpl(thisHost,
+                    Collections.emptySet(),
+                    Collections.emptySet(),
+                    Collections.emptySet(),
+                    Collections.emptySet()
            ));


nit: it should be 4 not 8 spaces

Suggested change

localMetadata.set(new StreamsMetadataImpl(thisHost,

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet()

));

localMetadata.set(new StreamsMetadataImpl(

thisHost,

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet()

));

cadonna · 2021-06-24T11:25:53Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java

+                final StreamsMetadata metadata = new StreamsMetadataImpl(hostInfo,
+                        activeStoresOnHost,
+                        activePartitionsOnHost,
+                        standbyStoresOnHost,
+                        standbyPartitionsOnHost);


Suggested change

final StreamsMetadata metadata = new StreamsMetadataImpl(hostInfo,

activeStoresOnHost,

activePartitionsOnHost,

standbyStoresOnHost,

standbyPartitionsOnHost);

final StreamsMetadata metadata = new StreamsMetadataImpl(

hostInfo,

activeStoresOnHost,

activePartitionsOnHost,

standbyStoresOnHost,

standbyPartitionsOnHost

);

cadonna · 2021-06-24T11:43:27Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImpl.java

+        if (producerClientIds != null) {
+            this.producerClientIds = Collections.unmodifiableSet(producerClientIds);
+        } else {
+            this.producerClientIds = Collections.emptySet();
+        }


I could not reproduce this issue. the set of producer IDs should never be null. If this is the case the test is wrong.

cadonna · 2021-06-24T11:44:25Z

streams/src/main/java/org/apache/kafka/streams/state/internals/StreamsMetadataImpl.java

+    public final static StreamsMetadataImpl NOT_AVAILABLE = new StreamsMetadataImpl(HostInfo.unavailable(),
+            Collections.emptySet(),
+            Collections.emptySet(),
+            Collections.emptySet(),
+            Collections.emptySet());


Suggested change

public final static StreamsMetadataImpl NOT_AVAILABLE = new StreamsMetadataImpl(HostInfo.unavailable(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet());

public final static StreamsMetadataImpl NOT_AVAILABLE = new StreamsMetadataImpl(

HostInfo.unavailable(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet(),

Collections.emptySet()

);

cadonna · 2021-06-24T12:02:09Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImplTest.java

+    public static final TaskMetadata TM_0 = new TaskMetadataImpl(
+            TASK_ID_0,
+            mkSet(TP_0_0, TP_1_0),
+            mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),
+            mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),
+            Optional.of(3L));


nit:

Suggested change

public static final TaskMetadata TM_0 = new TaskMetadataImpl(

TASK_ID_0,

mkSet(TP_0_0, TP_1_0),

mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),

mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),

Optional.of(3L));

public static final TaskMetadata TM_0 = new TaskMetadataImpl(

TASK_ID_0,

mkSet(TP_0_0, TP_1_0),

mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),

mkMap(mkEntry(TP_0_0, 1L), mkEntry(TP_1_0, 2L)),

Optional.of(3L)

);

here and below.

cadonna · 2021-06-24T12:04:26Z

streams/src/main/java/org/apache/kafka/streams/ThreadMetadata.java

+
+    /**
+     * Compares the specified object with this ThreadMetadata. Returns {@code true} if and only if the specified object is
+     * also a TaskMetadata and both {@code threadName()} are equal, {@code threadState()} are equal, {@code activeTasks()} contain the same


TaskMetadata -> ThreadMetadata

cadonna · 2021-06-24T12:09:28Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/ThreadMetadataImplTest.java

+    }
+
+    @Test
+    public void shouldFollowHashCodeAndEqualsContract() {


I prefer to have separate tests, because when something fails you immediately know the cause. But if you like to keep like that, I am fine with it.

cadonna · 2021-06-24T12:10:12Z

streams/src/test/java/org/apache/kafka/streams/state/StreamsMetadataTest.java

+        streamsMetadata = new StreamsMetadataImpl(
+                HOST_INFO,
+                STATE_STORE_NAMES,
+                TOPIC_PARTITIONS,
+                STAND_BY_STORE_NAMES,
+                STANDBY_TOPIC_PARTITIONS


nit: The indentation was actually correct before.

cadonna · 2021-06-24T12:10:30Z

streams/src/test/java/org/apache/kafka/streams/state/StreamsMetadataTest.java

@@ -55,6 +62,63 @@ public void shouldNotAllowModificationOfInternalStateViaGetters() {
        assertTrue(isUnmodifiable(streamsMetadata.standbyStateStoreNames()));
    }

+    @Test
+    public void shouldFollowHashCodeAndEqualsContract() {


see my other comment about separate tests.

jlprat · 2021-06-24T12:21:43Z

Hi @cadonna I'll try to fix address the comments either today or tomorrow.

Regarding the SourceConnectorsIntegrationTest failing in Jenkins, do you agree that is unrelated?

cadonna · 2021-06-24T14:06:52Z

Regarding the SourceConnectorsIntegrationTest failing in Jenkins, do you agree that is unrelated?

Yes, I agree!

It includes - Split equality tests by cases - Add JavaDoc on methods - Fix formatting - Use proper assertThat instead of other means - Fix names, use the proper names consistently

jlprat · 2021-06-25T10:45:58Z

Hi @cadonna I applied all the feedback you provided. I gave my best at adding some Extra sentences on the JavaDocs for the interfaces. Let me know if they suffice.

cadonna

@jlprat I did a pass over the javadocs.

streams/src/main/java/org/apache/kafka/streams/StreamsMetadata.java

streams/src/main/java/org/apache/kafka/streams/ThreadMetadata.java

Co-authored-by: Bruno Cadonna <cadonna@apache.org>

jlprat · 2021-06-25T13:49:00Z

@cadonna accepted your suggestions. Thanks a bunch!a

cadonna

@jlprat, Thanks a lot!

LGTM!

cadonna · 2021-06-25T15:52:20Z

I am just waiting for the builds to finish and then I will merge.

jlprat · 2021-06-25T16:20:56Z

It seems that the test failures were RaftClusterTest and SslTransportLayerTest ones. Both groups being known flaky ones AFAIU.

cadonna · 2021-06-25T16:29:21Z

The test failures are unrelated and known to be flaky:

Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[1] tlsProtocol=TLSv1.2, useInlinePem=false
Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[2] tlsProtocol=TLSv1.2, useInlinePem=true
Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[1] tlsProtocol=TLSv1.2, useInlinePem=false
Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[2] tlsProtocol=TLSv1.2, useInlinePem=true
Build / JDK 8 and Scala 2.12 / kafka.server.RaftClusterTest.testCreateClusterAndCreateAndManyTopics()
Build  / JDK 11 and Scala 2.13 /  org.apache.kafka.common.network.SslTransportLayerTest.[1]  tlsProtocol=TLSv1.2, useInlinePem=false
Build / JDK 11 and Scala 2.13 / org.apache.kafka.common.network.SslTransportLayerTest.[2] tlsProtocol=TLSv1.2, useInlinePem=true
Build  / JDK 11 and Scala 2.13 /  org.apache.kafka.common.network.SslTransportLayerTest.[3]  tlsProtocol=TLSv1.3, useInlinePem=false
Build  / JDK 11 and Scala 2.13 /  org.apache.kafka.common.network.SslTransportLayerTest.[1]  tlsProtocol=TLSv1.2, useInlinePem=false
Build / JDK 11 and Scala 2.13 / org.apache.kafka.common.network.SslTransportLayerTest.[2] tlsProtocol=TLSv1.2, useInlinePem=true
Build  / JDK 11 and Scala 2.13 /  org.apache.kafka.common.network.SslTransportLayerTest.[3]  tlsProtocol=TLSv1.3, useInlinePem=false
Build / JDK 11 and Scala 2.13 / kafka.server.RaftClusterTest.testCreateClusterAndCreateAndManyTopicsWithManyPartitions()
Build / JDK 11 and Scala 2.13 / kafka.server.RaftClusterTest.testCreateClusterAndCreateAndManyTopics()

jlprat · 2021-06-25T16:43:26Z

Thanks a lot @cadonna and @ableegoldman for the reviews!

…PI (apache#10840) Implementation of KIP-744. Creates new Interfaces for TaskMetadata, ThreadMetadata, and StreamsMetadata, providing internal implementations for each of them. Deprecates current TaskMetadata, ThreadMetadata under o.a.k.s.processor, and SreamsMetadata under a.o.k.s.state. Updates references on internal classes from deprecated classes to new interfaces. Deprecates methods on KafkaStreams returning deprecated ThreadMeatada and StreamsMetadata, and provides new ones returning the new interfaces. Update Javadocs referencing to deprecated classes and methods to point to the right ones. Co-authored-by: Bruno Cadonna <cadonna@apache.org> Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Bruno Cadonna <cadonna@apache.org>

jlprat commented Jun 8, 2021

View reviewed changes

ableegoldman reviewed Jun 23, 2021

View reviewed changes

cadonna reviewed Jun 23, 2021

View reviewed changes

jlprat force-pushed the KAFKA-12849-TaskMetadata-ThreadMetadata-StreamsMetadata-to-Interface branch from d266b98 to 9c3ffe3 Compare June 23, 2021 13:32

jlprat added 4 commits June 23, 2021 15:38

Applies suggestions regarding method names, refactors, and formatting

1945750

Add tests for equals hashcode and unmodifiable collections

a419c89

Add guard against null otherwise StreamThreadTest tests were failing

a6a0bec

jlprat force-pushed the KAFKA-12849-TaskMetadata-ThreadMetadata-StreamsMetadata-to-Interface branch from 9c3ffe3 to a6a0bec Compare June 23, 2021 14:03

jlprat commented Jun 23, 2021

View reviewed changes

cadonna reviewed Jun 24, 2021

View reviewed changes

Applies feedback

4403e8c

It includes - Split equality tests by cases - Add JavaDoc on methods - Fix formatting - Use proper assertThat instead of other means - Fix names, use the proper names consistently

jlprat requested a review from cadonna June 25, 2021 10:47

cadonna reviewed Jun 25, 2021

View reviewed changes

Apply suggestions from code review

8defa04

Co-authored-by: Bruno Cadonna <cadonna@apache.org>

cadonna approved these changes Jun 25, 2021

View reviewed changes

cadonna merged commit 6655a09 into apache:trunk Jun 25, 2021

jlprat deleted the KAFKA-12849-TaskMetadata-ThreadMetadata-StreamsMetadata-to-Interface branch June 25, 2021 16:43

	* @deprecated since 3.0, not intended for public use, use {@link org.apache.kafka.streams.TaskMetadata} instead.
	* @deprecated since 3.0, use {@link org.apache.kafka.streams.TaskMetadata} instead.

	<code>org.apache.kafka.streams.TaskMetadata</code> which offers these better methods, for example, by using the new <code>ThreadMetadata#getActiveTasks</code> and <code>ThreadMetadata#getStandbyTasks</code>.
	<code>org.apache.kafka.streams.TaskMetadata</code> which offers these better methods, for example, by using the new <code>ThreadMetadata#activeTasks</code> and <code>ThreadMetadata#standbyTasks</code>.

		*/
		@Deprecated

KAFKA-12849: KIP-744 TaskMetadata ThreadMetadata StreamsMetadata as API #10840

KAFKA-12849: KIP-744 TaskMetadata ThreadMetadata StreamsMetadata as API #10840

Conversation

jlprat commented Jun 8, 2021

Committer Checklist (excluded from commit message)

jlprat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat commented Jun 8, 2021

jlprat commented Jun 8, 2021

jlprat commented Jun 15, 2021

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat commented Jun 23, 2021

jlprat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat commented Jun 23, 2021

jlprat commented Jun 24, 2021

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat commented Jun 24, 2021

cadonna commented Jun 24, 2021

jlprat commented Jun 25, 2021

cadonna left a comment

Choose a reason for hiding this comment

jlprat commented Jun 25, 2021

cadonna left a comment

Choose a reason for hiding this comment

cadonna commented Jun 25, 2021

jlprat commented Jun 25, 2021

cadonna commented Jun 25, 2021