vivek807 commented on code in PR #19056:
URL: https://github.com/apache/druid/pull/19056#discussion_r2883130960
##########
extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java:
##########
@@ -252,45 +253,43 @@ private void watch()
private void keepWatching(String labelSelector, String resourceVersion)
{
String nextResourceVersion = resourceVersion;
+
while (lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) {
- try {
- WatchResult iter =
- k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector,
nextResourceVersion, nodeRole);
+ try (WatchResult iter =
k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector,
nextResourceVersion, nodeRole)) {
if (iter == null) {
// history not available, we need to start from scratch
return;
}
- try {
- while (iter.hasNext()) {
- Watch.Response<DiscoveryDruidNodeAndResourceVersion> item =
iter.next();
- if (item != null && item.type != null && item.object != null) {
- switch (item.type) {
- case WatchResult.ADDED:
- baseNodeRoleWatcher.childAdded(item.object.getNode());
- break;
- case WatchResult.DELETED:
- baseNodeRoleWatcher.childRemoved(item.object.getNode());
- break;
- default:
- }
-
- // This should be updated after the action has been dealt with
successfully
- nextResourceVersion = item.object.getResourceVersion();
-
- } else {
- // Try again by starting the watch from the beginning. This
can happen if the
- // watch goes bad.
- LOGGER.debug("Received NULL item while watching role[%s].
Restarting watch.", this.nodeRole);
- return;
- }
+ while (iter.hasNext()) {
+ Watch.Response<DiscoveryDruidNodeAndResourceVersion> item =
iter.next();
+
+ if (item == null || item.type == null || item.object == null) {
+ LOGGER.debug("Received NULL item while watching role[%s].
Restarting watch.", this.nodeRole);
+ return;
}
- }
- finally {
- iter.close();
+
+ switch (item.type) {
+ case WatchResult.ADDED:
+ baseNodeRoleWatcher.childAdded(item.object.getNode());
+ break;
+ case WatchResult.DELETED:
+ baseNodeRoleWatcher.childRemoved(item.object.getNode());
+ break;
+ default:
+ }
+
+ // This should be updated after the action has been dealt with
successfully
+ nextResourceVersion = item.object.getResourceVersion();
}
+ LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole);
+ return;
+ }
+ catch (StreamResetException ex) {
+ LOGGER.debug("Watch stream terminated normally for role[%s],
restarting", this.nodeRole);
+ return;
}
Review Comment:
Added test
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
