Skip to content

Commit adf0fc8

Browse files
authored
Report each plugin once in worker heartbeats (#3150)
1 parent c424910 commit adf0fc8

3 files changed

Lines changed: 57 additions & 4 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,8 @@ to docs, or any other relevant information.
7979
behavior during replay.
8080
- Test server now honors retry expiration deadlines that fall exactly on a whole second. Previously such deadlines were
8181
ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`.
82+
- Worker heartbeats now list each plugin name once. Previously a plugin registered on both the
83+
`WorkflowClient` and the `WorkerFactory` was listed twice.
8284

8385
## Previous releases
8486

‎temporal-sdk/src/main/java/io/temporal/worker/Worker.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -775,10 +775,12 @@ Supplier<WorkerHeartbeat> buildHeartbeatCallback(
775775
hb.setTotalStickyCacheMiss(cache.getCacheMisses());
776776
hb.setCurrentStickyCacheSize(cache.getCurrentCacheSize());
777777

778-
// Plugins
779-
for (WorkerPlugin plugin : plugins) {
780-
hb.addPlugins(PluginInfo.newBuilder().setName(plugin.getName()).build());
781-
}
778+
// A plugin can be registered on both the client and the worker factory, so report each
779+
// name once.
780+
plugins.stream()
781+
.map(WorkerPlugin::getName)
782+
.distinct()
783+
.forEach(name -> hb.addPlugins(PluginInfo.newBuilder().setName(name).build()));
782784

783785
return hb.build();
784786
}
Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package io.temporal.worker;
2+
3+
import static org.junit.Assert.assertEquals;
4+
5+
import io.temporal.api.worker.v1.PluginInfo;
6+
import io.temporal.client.WorkflowClient;
7+
import io.temporal.client.WorkflowClientOptions;
8+
import io.temporal.common.SimplePlugin;
9+
import io.temporal.testing.CloudTestExclusion.RequiresLocalServer;
10+
import io.temporal.testing.CloudTestExclusionNote;
11+
import io.temporal.testing.TestWorkflowEnvironment;
12+
import java.util.Collections;
13+
import java.util.List;
14+
import java.util.stream.Collectors;
15+
import org.junit.Test;
16+
import org.junit.experimental.categories.Category;
17+
18+
@CloudTestExclusionNote("This test directly creates and controls a local test service.")
19+
@Category(RequiresLocalServer.class)
20+
public class WorkerHeartbeatPluginsTest {
21+
22+
@Test
23+
public void pluginRegisteredOnClientAndFactoryIsReportedOnce() {
24+
SimplePlugin plugin = SimplePlugin.newBuilder("test-plugin").build();
25+
26+
try (TestWorkflowEnvironment env = TestWorkflowEnvironment.newInstance()) {
27+
WorkflowClient client =
28+
WorkflowClient.newInstance(
29+
env.getWorkflowServiceStubs(),
30+
WorkflowClientOptions.newBuilder()
31+
.setNamespace(env.getNamespace())
32+
.setPlugins(plugin)
33+
.build());
34+
WorkerFactory factory =
35+
WorkerFactory.newInstance(
36+
client, WorkerFactoryOptions.newBuilder().setPlugins(plugin).build());
37+
Worker worker = factory.newWorker("test-queue");
38+
factory.start();
39+
40+
List<String> reportedPlugins =
41+
worker.buildHeartbeatCallback("test-worker-group", null).get().getPluginsList().stream()
42+
.map(PluginInfo::getName)
43+
.collect(Collectors.toList());
44+
assertEquals(Collections.singletonList("test-plugin"), reportedPlugins);
45+
46+
factory.shutdownNow();
47+
}
48+
}
49+
}

0 commit comments

Comments
 (0)