#445

Committed 24 Jul 2025 02:27PM UTC coverage: 69.257% (-7.1%) from 76.395%

Build # #445

Build Type

push

Committed by

Commit Message

Merge branch 'hive-3.x' of github.com:ExpediaGroup/waggle-dance into hive-3.x

Run Details

2620 of 3783 relevant lines covered (69.26%)

0.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.57

/waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/server/TTransportMonitor.java

/**
 * Copyright (C) 2016-2025 Expedia, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hotels.bdp.waggledance.server;

import java.io.Closeable;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import javax.annotation.PreDestroy;
import javax.annotation.WillClose;

import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.AllArgsConstructor;

import com.google.common.annotations.VisibleForTesting;

import com.hotels.bdp.waggledance.conf.WaggleDanceConfiguration;

@Component
public class TTransportMonitor {
  private static Logger log = LoggerFactory.getLogger(TTransportMonitor.class);
  static final String METRIC_NAME_OPEN_TRANSPORTS = "com_hotels_bdp_waggledance_open_transports_gauge";
  private static final Logger LOG = LoggerFactory.getLogger(TTransportMonitor.class);

  @AllArgsConstructor
  private static class ActionContainer {
    private final TTransport transport;
    private final Closeable action;
  }

  private final ScheduledExecutorService scheduler;
  private final ConcurrentLinkedQueue<ActionContainer> transports = new ConcurrentLinkedQueue<>();

  @Autowired
  public TTransportMonitor(WaggleDanceConfiguration waggleDanceConfiguration, MeterRegistry meterRegistry) {
    this(waggleDanceConfiguration, Executors.newScheduledThreadPool(1), meterRegistry);
  }

  @VisibleForTesting
  TTransportMonitor(
      WaggleDanceConfiguration waggleDanceConfiguration,
      ScheduledExecutorService scheduler,
      MeterRegistry meterRegistry) {
    this.scheduler = scheduler;
    Gauge.builder(METRIC_NAME_OPEN_TRANSPORTS, transports, ConcurrentLinkedQueue::size).register(meterRegistry);
    Runnable monitor = () -> {
      log.debug("Releasing disconnected sessions");
      Iterator<ActionContainer> iterator = transports.iterator();
      while (iterator.hasNext()) {
        ActionContainer actionContainer = iterator.next();
        if (actionContainer.transport.peek()) {
          continue;
        }
        try {
          actionContainer.action.close();
        } catch (Exception e) {
          log.warn("Error closing action", e);
        }
        try {
          actionContainer.transport.close();
        } catch (Exception e) {
          log.warn("Error closing transport", e);
        }
        iterator.remove();
      }
      LOG.info("Number of open transports (#connections clients -> WD ): {}", transports.size());
    };
    this.scheduler
        .scheduleAtFixedRate(monitor, waggleDanceConfiguration.getDisconnectConnectionDelay(),
            waggleDanceConfiguration.getDisconnectConnectionDelay(), waggleDanceConfiguration.getDisconnectTimeUnit());
  }

  @PreDestroy
  public void shutdown() {
    scheduler.shutdown();
  }

  public void monitor(@WillClose TTransport transport, @WillClose Closeable action) {
    transports.offer(new ActionContainer(transport, action));
  }

}

1	/**
2	* Copyright (C) 2016-2025 Expedia, Inc.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package com.hotels.bdp.waggledance.server;
17
18	import java.io.Closeable;
19	import java.util.Iterator;
20	import java.util.concurrent.ConcurrentLinkedQueue;
21	import java.util.concurrent.Executors;
22	import java.util.concurrent.ScheduledExecutorService;
23
24	import javax.annotation.PreDestroy;
25	import javax.annotation.WillClose;
26
27	import org.apache.thrift.transport.TTransport;
28	import org.slf4j.Logger;
29	import org.slf4j.LoggerFactory;
30	import org.springframework.beans.factory.annotation.Autowired;
31	import org.springframework.stereotype.Component;
32
33	import io.micrometer.core.instrument.Gauge;
34	import io.micrometer.core.instrument.MeterRegistry;
35	import lombok.AllArgsConstructor;
36
37	import com.google.common.annotations.VisibleForTesting;
38
39	import com.hotels.bdp.waggledance.conf.WaggleDanceConfiguration;
40
41	@Component
42	public class TTransportMonitor {
43	private static Logger log = LoggerFactory.getLogger(TTransportMonitor.class);	1✔
44	static final String METRIC_NAME_OPEN_TRANSPORTS = "com_hotels_bdp_waggledance_open_transports_gauge";
45	private static final Logger LOG = LoggerFactory.getLogger(TTransportMonitor.class);	1✔
46
47	@AllArgsConstructor
48	private static class ActionContainer {
49	private final TTransport transport;
50	private final Closeable action;
51	}
52
53	private final ScheduledExecutorService scheduler;
54	private final ConcurrentLinkedQueue<ActionContainer> transports = new ConcurrentLinkedQueue<>();	1✔
55
56	@Autowired
57	public TTransportMonitor(WaggleDanceConfiguration waggleDanceConfiguration, MeterRegistry meterRegistry) {
58	this(waggleDanceConfiguration, Executors.newScheduledThreadPool(1), meterRegistry);	×
59	}	×
60
61	@VisibleForTesting
62	TTransportMonitor(
63	WaggleDanceConfiguration waggleDanceConfiguration,
64	ScheduledExecutorService scheduler,
65	MeterRegistry meterRegistry) {	1✔
66	this.scheduler = scheduler;	1✔
67	Gauge.builder(METRIC_NAME_OPEN_TRANSPORTS, transports, ConcurrentLinkedQueue::size).register(meterRegistry);	1✔
68	Runnable monitor = () -> {	1✔
69	log.debug("Releasing disconnected sessions");	1✔
70	Iterator<ActionContainer> iterator = transports.iterator();	1✔
71	while (iterator.hasNext()) {	1✔
72	ActionContainer actionContainer = iterator.next();	1✔
73	if (actionContainer.transport.peek()) {	1✔
74	continue;	1✔
75	}
76	try {
77	actionContainer.action.close();	1✔
78	} catch (Exception e) {	1✔
79	log.warn("Error closing action", e);	1✔
80	}	1✔
81	try {
82	actionContainer.transport.close();	1✔
83	} catch (Exception e) {	1✔
84	log.warn("Error closing transport", e);	1✔
85	}	1✔
86	iterator.remove();	1✔
87	}	1✔
88	LOG.info("Number of open transports (#connections clients -> WD ): {}", transports.size());	1✔
89	};	1✔
90	this.scheduler	1✔
91	.scheduleAtFixedRate(monitor, waggleDanceConfiguration.getDisconnectConnectionDelay(),	1✔
92	waggleDanceConfiguration.getDisconnectConnectionDelay(), waggleDanceConfiguration.getDisconnectTimeUnit());	1✔
93	}	1✔
94
95	@PreDestroy
96	public void shutdown() {
97	scheduler.shutdown();	×
98	}	×
99
100	public void monitor(@WillClose TTransport transport, @WillClose Closeable action) {
101	transports.offer(new ActionContainer(transport, action));	1✔
102	}	1✔
103
104	}

HotelsDotCom / waggle-dance / #445

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous