NSAmelchev commented on a change in pull request #73:
URL: https://github.com/apache/ignite-extensions/pull/73#discussion_r724120135
##########
File path:
modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java
##########
@@ -408,4 +438,57 @@ private boolean checkFuts(boolean res,
List<IgniteInternalFuture<?>> futs) {
/** */
protected abstract List<IgniteInternalFuture<?>> startActiveActiveCdc();
+
+ /** */
+ protected abstract void checkConsumerMetrics(Function<String, Long>
longMetric);
+
+ /** */
+ protected void checkMetrics() {
+ for (int i = 0; i < cdcs.size(); i++) {
+ IgniteConfiguration cfg = getFieldValue(cdcs.get(i), "igniteCfg");
+
+ MetricRegistry mreg = getFieldValue(cdcs.get(i), "mreg");
+
+ assertNotNull(mreg);
+
+ checkMetrics(
+ m -> mreg.<LongMetric>findMetric(m).value(),
+ m -> mreg.<ObjectMetric<String>>findMetric(m).value()
+ );
+
+ Function<DynamicMBean, Function<String, ?>> jmxVal = mxBean -> m
-> {
+ try {
+ return mxBean.getAttribute(m);
+ }
+ catch (Exception e) {
+ throw new IgniteException(e);
+ }
+ };
+
+ DynamicMBean jmxCdcReg =
metricRegistry(cdcInstanceName(cfg.getIgniteInstanceName()), null, "cdc");
+
+ checkMetrics((Function<String, Long>)jmxVal.apply(jmxCdcReg),
(Function<String, String>)jmxVal.apply(jmxCdcReg));
+
+ DynamicMBean jmxConsumerReg =
metricRegistry(cdcInstanceName(cfg.getIgniteInstanceName()), "cdc", "consumer");
+
+ checkConsumerMetrics((Function<String,
Long>)jmxVal.apply(jmxConsumerReg));
+ }
+ }
+
+ /** */
+ private void checkMetrics(Function<String, Long> longMetric,
Function<String, String> strMetric) {
+ long committedSegIdx = longMetric.apply(COMMITTED_SEG_IDX);
+ long curSegIdx = longMetric.apply(CUR_SEG_IDX);
+
+ assertTrue(committedSegIdx <= curSegIdx);
+
+ assertTrue(longMetric.apply(COMMITTED_SEG_OFFSET) >= 0);
+ assertTrue(longMetric.apply(LAST_SEG_CONSUMPTION_TIME) > 0);
+
+ for (String m : new String[] {BINARY_META_DIR, MARSHALLER_DIR,
CDC_DIR})
+ assertTrue(new File(strMetric.apply(m)).exists());
+
+ assertNotNull(longMetric.apply(LAST_EVT_TIME));
Review comment:
Can we check that the metric values are positive?
(`checkConsumerMetrics` too)
##########
File path:
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java
##########
@@ -72,9 +68,11 @@ public CdcEventsApplier(int maxBatchSize) {
* @param evts Events to process.
* @throws IgniteCheckedException If failed.
Review comment:
Add the `@return` description please
##########
File path:
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java
##########
@@ -137,17 +135,21 @@ protected void apply(Iterable<CdcEvent> evts) throws
IgniteCheckedException {
* @param applyRmv Apply remove batch flag supplier.
* @throws IgniteCheckedException In case of error.
Review comment:
Add the `@return` description please
##########
File path:
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java
##########
@@ -83,26 +103,34 @@ public IgniteToIgniteCdcStreamer(IgniteConfiguration
destIgniteCfg, boolean only
}
/** {@inheritDoc} */
- @Override public void start() {
+ @Override public void start(MetricRegistry mreg) {
if (log.isInfoEnabled())
log.info("Ignite To Ignite Streamer [cacheIds=" + cachesIds + ']');
dest = (IgniteEx)Ignition.start(destIgniteCfg);
+
+ this.evtsCnt = mreg.longMetric(EVTS_CNT, EVTS_CNT_DESC);
+ this.lastEvtTs = mreg.longMetric(LAST_EVT_TIME, LAST_EVT_TIME_DESC);
}
/** {@inheritDoc} */
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
try {
- apply(() -> F.iterator(
+ long msgsSnt0 = apply(() -> F.iterator(
Review comment:
Lets rename to `msgsSnt` ?
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
