You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "parveensania (via GitHub)" <gi...@apache.org> on 2026/04/20 18:39:08 UTC
[PR] Add connectivity type tracking to endpoint consumption logic [beam]
parveensania opened a new pull request, #38249:
URL: https://github.com/apache/beam/pull/38249
This PR updates the endpoint consumption logic in FanOutStreamingEngineWorkerHarness to trigger refreshes whenever the connectivity type changes, even if the metadata version remains the same.
Previously, the harness only refreshed endpoints if the metadata version increased. This change ensures the client immediately picks new endpoint types if backend settings change, without waiting for a version increment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3128040623
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -271,8 +279,14 @@ private void consumeWorkerMetadata(WindmillEndpoints windmillEndpoints) {
synchronized (metadataLock) {
// Only process versions greater than what we currently have to prevent double processing of
// metadata. workerMetadataConsumer is single-threaded so we maintain ordering.
- if (windmillEndpoints.version() > pendingMetadataVersion) {
+ // The endpoints are also consumed if the version is the same but the type of endpoints
+ // sent by the server has changed.
+ if (windmillEndpoints.version() > pendingMetadataVersion
+ || (windmillEndpoints.version() == pendingMetadataVersion
Review Comment:
Done. Considering we only expect monotonically increasing version, we can remove this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121408766
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/WindmillEndpoints.java:
##########
@@ -39,13 +39,34 @@
*/
@AutoValue
public abstract class WindmillEndpoints {
+ public enum Type {
+ UNKNOWN,
+ CLOUDPATH,
+ BORG,
Review Comment:
done
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -106,6 +106,9 @@ public final class FanOutStreamingEngineWorkerHarness implements StreamingWorker
@GuardedBy("this")
private long activeMetadataVersion;
+ @GuardedBy("metadataLock")
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #38249:
URL: https://github.com/apache/beam/pull/38249#issuecomment-4301590931
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/38249?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `75.00000%` with `3 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.52%. Comparing base ([`79fb360`](https://app.codecov.io/gh/apache/beam/commit/79fb3600f9712e2eea962941c65d4812b3121445?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c849e1b`](https://app.codecov.io/gh/apache/beam/commit/c849e1b904ce1936890684fe9aa26488631171ca?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 8 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/38249?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ng/harness/FanOutStreamingEngineWorkerHarness.java](https://app.codecov.io/gh/apache/beam/pull/38249?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2Fstreaming%2Fharness%2FFanOutStreamingEngineWorkerHarness.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvc3RyZWFtaW5nL2hhcm5lc3MvRmFuT3V0U3RyZWFtaW5nRW5naW5lV29ya2VySGFybmVzcy5qYXZh) | 75.00% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/38249?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #38249 +/- ##
=============================================
+ Coverage 54.62% 58.52% +3.90%
- Complexity 1689 15430 +13741
=============================================
Files 1067 2851 +1784
Lines 168155 280096 +111941
Branches 1226 12340 +11114
=============================================
+ Hits 91853 163934 +72081
- Misses 74108 109747 +35639
- Partials 2194 6415 +4221
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/38249/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/38249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `64.61% <75.00%> (-2.74%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/38249?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38249:
URL: https://github.com/apache/beam/pull/38249#issuecomment-4283388979
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request enhances the endpoint consumption logic within the streaming engine harness to be more responsive to backend connectivity changes. By tracking the connectivity type alongside the metadata version, the system can now proactively refresh endpoints when backend settings change, ensuring better connectivity management without relying solely on metadata version increments.
### Highlights
* **Connectivity Type Tracking**: Introduced a new 'Type' enum in WindmillEndpoints to track connectivity types (e.g., CLOUDPATH, BORG, DIRECTPATH) and updated the endpoint consumption logic to trigger refreshes when this type changes, even if the metadata version remains the same.
* **Refreshed Consumption Logic**: Modified FanOutStreamingEngineWorkerHarness to track the active metadata type and ensure that endpoint updates are processed if either the version increases or the connectivity type changes.
* **Stream Metadata Handling**: Updated GrpcGetWorkerMetadataStream to remove the stale metadata check, allowing the harness to handle all incoming metadata updates and determine if they should be consumed based on the new logic.
🧠 **New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121415932
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -271,7 +275,11 @@ private void consumeWorkerMetadata(WindmillEndpoints windmillEndpoints) {
synchronized (metadataLock) {
// Only process versions greater than what we currently have to prevent double processing of
// metadata. workerMetadataConsumer is single-threaded so we maintain ordering.
- if (windmillEndpoints.version() > pendingMetadataVersion) {
+ // The endpoints are also consumed if the version is the same but the type of endpoints
+ // sent by the server has changed.
+ if (windmillEndpoints.version() > pendingMetadataVersion
+ || (windmillEndpoints.version() == pendingMetadataVersion
+ && windmillEndpoints.type() != activeMetadataType)) {
Review Comment:
The reason I didn't think we need to sync on pendingMetadataType is because I don't expect endpoint type to change often. It should be a rare event triggered manually.
Added check on pendingMetadataType if we think this will help in future proofing logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121436449
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -271,8 +279,14 @@ private void consumeWorkerMetadata(WindmillEndpoints windmillEndpoints) {
synchronized (metadataLock) {
// Only process versions greater than what we currently have to prevent double processing of
// metadata. workerMetadataConsumer is single-threaded so we maintain ordering.
- if (windmillEndpoints.version() > pendingMetadataVersion) {
+ // The endpoints are also consumed if the version is the same but the type of endpoints
+ // sent by the server has changed.
+ if (windmillEndpoints.version() > pendingMetadataVersion
+ || (windmillEndpoints.version() == pendingMetadataVersion
Review Comment:
do we need this version check? Seems like it is enough to check if metadata changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121410031
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -288,11 +296,17 @@ private synchronized void consumeWindmillWorkerEndpoints(WindmillEndpoints newWi
}
}
- LOG.debug(
- "Consuming new endpoints: {}. previous metadata version: {}, current metadata version: {}",
+ WindmillEndpoints.Type previousType;
+ synchronized (metadataLock) {
Review Comment:
synchronization on metadataLock no longer needed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121410614
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarnessTest.java:
##########
@@ -348,6 +348,48 @@ public void testOnNewWorkerMetadata_redistributesBudget() throws InterruptedExce
TimeUnit.SECONDS.sleep(WAIT_FOR_METADATA_INJECTIONS_SECONDS);
}
+ @Test
+ public void testOnNewWorkerMetadata_consumesEndpointsOnConnectivityTypeChange()
Review Comment:
updated the test case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "parveensania (via GitHub)" <gi...@apache.org>.
parveensania commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3121411238
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarnessTest.java:
##########
@@ -348,6 +348,48 @@ public void testOnNewWorkerMetadata_redistributesBudget() throws InterruptedExce
TimeUnit.SECONDS.sleep(WAIT_FOR_METADATA_INJECTIONS_SECONDS);
}
+ @Test
+ public void testOnNewWorkerMetadata_consumesEndpointsOnConnectivityTypeChange()
+ throws InterruptedException {
+ String workerToken = "workerToken1";
+
+ WorkerMetadataResponse firstWorkerMetadata =
+ WorkerMetadataResponse.newBuilder()
+ .setMetadataVersion(1)
+ .setEndpointType(Windmill.WorkerMetadataResponse.EndpointType.CLOUDPATH)
+ .addWorkEndpoints(
+ WorkerMetadataResponse.Endpoint.newBuilder()
+ .setBackendWorkerToken(workerToken)
+ .build())
+ .putAllGlobalDataEndpoints(DEFAULT)
+ .build();
+ WorkerMetadataResponse secondWorkerMetadata =
+ WorkerMetadataResponse.newBuilder()
+ .setMetadataVersion(1)
+ .setEndpointType(Windmill.WorkerMetadataResponse.EndpointType.DIRECTPATH)
+ .addWorkEndpoints(
+ WorkerMetadataResponse.Endpoint.newBuilder()
+ .setBackendWorkerToken(workerToken)
+ .build())
+ .putAllGlobalDataEndpoints(DEFAULT)
+ .build();
+
+ TestGetWorkBudgetDistributor getWorkBudgetDistributor = spy(new TestGetWorkBudgetDistributor());
+ fanOutStreamingEngineWorkProvider =
+ newFanOutStreamingEngineWorkerHarness(
+ GetWorkBudget.builder().setItems(1).setBytes(1).build(),
+ getWorkBudgetDistributor,
+ noOpProcessWorkItemFn());
+
+ fakeGetWorkerMetadataStub.injectWorkerMetadata(firstWorkerMetadata);
Review Comment:
added additional checks in testcase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz merged PR #38249:
URL: https://github.com/apache/beam/pull/38249
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38249:
URL: https://github.com/apache/beam/pull/38249#discussion_r3114262604
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -288,11 +296,17 @@ private synchronized void consumeWindmillWorkerEndpoints(WindmillEndpoints newWi
}
}
- LOG.debug(
- "Consuming new endpoints: {}. previous metadata version: {}, current metadata version: {}",
+ WindmillEndpoints.Type previousType;
+ synchronized (metadataLock) {
Review Comment:
merge it with the synchronized block in line 293?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarnessTest.java:
##########
@@ -348,6 +348,48 @@ public void testOnNewWorkerMetadata_redistributesBudget() throws InterruptedExce
TimeUnit.SECONDS.sleep(WAIT_FOR_METADATA_INJECTIONS_SECONDS);
}
+ @Test
+ public void testOnNewWorkerMetadata_consumesEndpointsOnConnectivityTypeChange()
Review Comment:
can you add a test case like testOnNewWorkerMetadata_correctlyRemovesStaleWindmillServers for cloudpath -> directpath -> cloudpath and directpath -> cloudpath -> directpath?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -271,7 +275,11 @@ private void consumeWorkerMetadata(WindmillEndpoints windmillEndpoints) {
synchronized (metadataLock) {
// Only process versions greater than what we currently have to prevent double processing of
// metadata. workerMetadataConsumer is single-threaded so we maintain ordering.
- if (windmillEndpoints.version() > pendingMetadataVersion) {
+ // The endpoints are also consumed if the version is the same but the type of endpoints
+ // sent by the server has changed.
+ if (windmillEndpoints.version() > pendingMetadataVersion
+ || (windmillEndpoints.version() == pendingMetadataVersion
+ && windmillEndpoints.type() != activeMetadataType)) {
Review Comment:
i think we need to check against a `pendingMetadataType` like `pendingMetadataVersion`
Consider a case the following case,
1. active is cloudpath
1. we receive a directpath config and enter here since the types don't match
1. we receive a cloudpath config, before the activeMetadataType is modified to directpath, now the check against activeMetadataType will prevent the cloudpath config from taking effect.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarnessTest.java:
##########
@@ -348,6 +348,48 @@ public void testOnNewWorkerMetadata_redistributesBudget() throws InterruptedExce
TimeUnit.SECONDS.sleep(WAIT_FOR_METADATA_INJECTIONS_SECONDS);
}
+ @Test
+ public void testOnNewWorkerMetadata_consumesEndpointsOnConnectivityTypeChange()
+ throws InterruptedException {
+ String workerToken = "workerToken1";
+
+ WorkerMetadataResponse firstWorkerMetadata =
+ WorkerMetadataResponse.newBuilder()
+ .setMetadataVersion(1)
+ .setEndpointType(Windmill.WorkerMetadataResponse.EndpointType.CLOUDPATH)
+ .addWorkEndpoints(
+ WorkerMetadataResponse.Endpoint.newBuilder()
+ .setBackendWorkerToken(workerToken)
+ .build())
+ .putAllGlobalDataEndpoints(DEFAULT)
+ .build();
+ WorkerMetadataResponse secondWorkerMetadata =
+ WorkerMetadataResponse.newBuilder()
+ .setMetadataVersion(1)
+ .setEndpointType(Windmill.WorkerMetadataResponse.EndpointType.DIRECTPATH)
+ .addWorkEndpoints(
+ WorkerMetadataResponse.Endpoint.newBuilder()
+ .setBackendWorkerToken(workerToken)
+ .build())
+ .putAllGlobalDataEndpoints(DEFAULT)
+ .build();
+
+ TestGetWorkBudgetDistributor getWorkBudgetDistributor = spy(new TestGetWorkBudgetDistributor());
+ fanOutStreamingEngineWorkProvider =
+ newFanOutStreamingEngineWorkerHarness(
+ GetWorkBudget.builder().setItems(1).setBytes(1).build(),
+ getWorkBudgetDistributor,
+ noOpProcessWorkItemFn());
+
+ fakeGetWorkerMetadataStub.injectWorkerMetadata(firstWorkerMetadata);
Review Comment:
check if `fanOutStreamingEngineWorkProvider.currentBackends()` has expected values?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/WindmillEndpoints.java:
##########
@@ -39,13 +39,34 @@
*/
@AutoValue
public abstract class WindmillEndpoints {
+ public enum Type {
+ UNKNOWN,
+ CLOUDPATH,
+ BORG,
Review Comment:
remove?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -106,6 +106,9 @@ public final class FanOutStreamingEngineWorkerHarness implements StreamingWorker
@GuardedBy("this")
private long activeMetadataVersion;
+ @GuardedBy("metadataLock")
Review Comment:
can change this to be guarded by `this` after adding a pendingMetaDataType guarded by `metadataLock`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add connectivity type tracking to endpoint consumption logic [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38249:
URL: https://github.com/apache/beam/pull/38249#issuecomment-4283847692
Assigning reviewers:
R: @tvalentyn added as fallback since no labels match configuration
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org