[BEAM-6161] Introduce PCollectionConsumerRegistry and add ElementCoun… #7272
Conversation
ajamato
force-pushed
the
element_count_dec14
branch
3 times, most recently
from
1069654
to
de52347
Compare
ajamato
changed the title
|
@swegner @Ardagan @robertwb @echauchot |
| @@ -146,8 +148,19 @@ public MetricUpdates getUpdates() { | |||
|
|
|||
| for (MetricUpdate<Long> mu : mus.counterUpdates()) { | |||
There was a problem hiding this comment.
mu, as well as mus are hard to read. Use monitoringInfo and monitoringInfos.
e -> entry. (can be mixed with error, or other abbreviations)
| @@ -584,21 +584,28 @@ public void onProgress(ProcessBundleProgressResponse progress) {} | |||
| @Override | |||
| public void onCompleted(ProcessBundleResponse response) { | |||
| // Assert the timestamps are non empty then 0 them out before comparing. | |||
| List<MonitoringInfo> actualMIs = new ArrayList<>(); | |||
| List<MonitoringInfo> actual = new ArrayList<MonitoringInfo>(); | |||
| builder = new SimpleMonitoringInfoBuilder(); | ||
| builder.setUrn(SimpleMonitoringInfoBuilder.ELEMENT_COUNT_URN); | ||
| builder.setPCollectionLabel("impulse.out"); | ||
| builder.setInt64Value(2); |
There was a problem hiding this comment.
Use unique values. Right now you have 3 places that use same values: element_count counter, userMetric counter and total amount of counters.
This might lead to misunderstanding.
| */ | ||
| package org.apache.beam.sdk.metrics; | ||
|
|
||
| public class LabelledMetrics { |
There was a problem hiding this comment.
Double checked this. It is difference between British and American English. I suggest we stick to American version here, since it is common for most of code.
| @@ -110,7 +110,7 @@ | |||
| } | |||
| Collection<FnDataReceiver<WindowedValue<OutputT>>> consumers = | |||
| (Collection) | |||
| pCollectionIdsToConsumers.get(getOnlyElement(pTransform.getOutputsMap().values())); | |||
| pCollectionConsumerRegistry.get(getOnlyElement(pTransform.getOutputsMap().values())); | |||
There was a problem hiding this comment.
Is there a reason why our outputs map contains only one element?
|
|
||
| /** | ||
| * A wrapping FnDataReceiverWindowedValue<T> which counts the number of elements consumed by the | ||
| * original nDataReceiverWindowedValue<T>. |
There was a problem hiding this comment.
missing F in FnDataReceiver...
| } | ||
|
|
||
| /** @return The Get the only FnDataReceiver for the pcollection. */ | ||
| public FnDataReceiver<WindowedValue<?>> getOnlyElement(String pCollectionId) { |
There was a problem hiding this comment.
This method is not used, please, remove. If I miss the usage, please rename to getReceiver, or getFnDataReceiver
| return pCollectionIdsToConsumers.keySet(); | ||
| } | ||
|
|
||
| /** @return The Get the only FnDataReceiver for the pcollection. */ |
There was a problem hiding this comment.
document IllegalArgumentException when multiple elements are returned for pCollectionId
|
|
||
| public MonitoringInfoMetricName(String urn, HashMap<String, String> labels) { | ||
| checkArgument(!Strings.isNullOrEmpty(urn), "MonitoringInfoMetricName urn must be non-empty"); | ||
| checkArgument(labels != null, "MonitoringInfoMetricName name must be non-null"); |
There was a problem hiding this comment.
Unclear comment.
Should state that labels parameter must not be null. If there are any required labels that should be present, it would be good to name those as well.
There was a problem hiding this comment.
Update comment, doing mroe would require adding a dep on SimpleMonitoringInfoBuilder, which would require adding a dep on runner-core-java-main. which is a backwards/looping dep. I'll add a TODO.
|
Please, add unittests for newly added classes. |
| Collection<FnDataReceiver<WindowedValue<KV<KeyT, AccumT>>>> consumers = | ||
| (Collection) | ||
| pCollectionConsumerRegistry.get( | ||
| //Collection<FnDataReceiver<WindowedValue<KV<KeyT, AccumT>>>> consumers = |
| @@ -56,6 +54,7 @@ | |||
| static class Factory<InputT> implements PTransformRunnerFactory<FlattenRunner<InputT>> { | |||
| @Override | |||
| public FlattenRunner<InputT> createRunnerForPTransform( | |||
| //public FlattenRunner<InputT> createRunnerForPTransform( | |||
| @@ -72,17 +71,16 @@ | |||
| throws IOException { | |||
|
|
|||
| // Give each input a MultiplexingFnDataReceiver to all outputs of the flatten. | |||
| ImmutableSet.Builder<FnDataReceiver<WindowedValue<InputT>>> consumersBuilder = | |||
| new ImmutableSet.Builder<>(); | |||
| // TODO ajamato: Somehow the consumersBuilder handles the casting for us before this change. | |||
| ElementCountFnDataReceiver wrappedConsumer = | ||
| pCollectionIdsToWrappedConsumer.getOrDefault(pCollectionId, null); | ||
| if (wrappedConsumer != null) { | ||
| throw new RuntimeException( |
There was a problem hiding this comment.
We might split this to builder and implementation. This will remove the option for user to invoke calls in wrong order.
There was a problem hiding this comment.
Interesting idea, but the builder would basically need to be the same as this. Having an exception thrown if built twice, or mroe things are added after, etc. Let's leave it as is for now, unless there is much more advantage
| new ElementCountFnDataReceiver<T>(consumer, pCollectionId); | ||
| pCollectionIdsToConsumers.put(pCollectionId, (FnDataReceiver) wrappedReceiver); | ||
| return wrappedReceiver; | ||
| public <T> void register(String pCollectionId, FnDataReceiver<WindowedValue<T>> consumer) { |
There was a problem hiding this comment.
Add documentation for method.
Note the order in which this method should be called regards getMultiplexingConsumer.
| @@ -0,0 +1,64 @@ | |||
| /* | |||
There was a problem hiding this comment.
this metrics package does not have tests for every file inside, so I don't want to add an unnecessary test here. This is just a refactoring of existing code. Its being tested
I don't see this test as adding much value because this code is already tested very well with the existing tests since this is a class used to implement the others in the package. Specifically MetricContainerImpl
| @@ -0,0 +1,56 @@ | |||
| /* | |||
ajamato
force-pushed
the
element_count_dec14
branch
from
ed15f51
to
a86310e
Compare
| // Represents a specific MonitoringInfo for a specific URN. | ||
| builder.setUrn(monitoringInfoName.getUrn()); | ||
| for (Entry<String, String> e : monitoringInfoName.getLabels().entrySet()) { | ||
| builder.setLabel(e.getKey(), e.getValue()); |
There was a problem hiding this comment.
would addLabel be a better name for this function?
| @ProcessElement | ||
| public void process(ProcessContext ctxt) { | ||
| Metrics.counter(RemoteExecutionTest.class, counterMetricName).inc(); | ||
| // Due to a bug impulse is producing two elements instead of one. | ||
| // So add this check to only emit these three elemenets. |
There was a problem hiding this comment.
| // So add this check to only emit these three elemenets. | |
| // So add this check to only emit these three elements. |
is there a JIRA link handy about this bug, that could be included here?
ajamato
force-pushed
the
element_count_dec14
branch
from
58f0191
to
d178db5
Compare
|
Seems to be an issue starting up the VMs, due to not getting the built dataflow container. E Handler for GET /v1.27/images/us.gcr.io/apache-beam-testing/java-postcommit-it/java:20190105011654/json returned error: No such image: us.gcr.io/apache-beam-testing/java-postcommit-it/java:20190105011654 |
|
retest this please |
|
|
||
| for (MetricUpdate<Long> mu : mus.counterUpdates()) { | ||
| for (MetricUpdate<Long> metricUpdate : metricUpdates.counterUpdates()) { | ||
| SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder(true); |
There was a problem hiding this comment.
Extract MonitoringInfo creation logic into a separate method.
| @@ -142,13 +143,25 @@ public MetricUpdates getUpdates() { | |||
| public Iterable<MonitoringInfo> getMonitoringInfos() { | |||
There was a problem hiding this comment.
Add unit test for this method
| /** Tests for {@link MonitoringInfoMetricName}. */ | ||
| public class MonitoringInfoMetricNameTest implements Serializable { | ||
|
|
||
| @Rule public final transient ExpectedException thrown = ExpectedException.none(); |
There was a problem hiding this comment.
rename to throws. Then throws.expect() will sound more natural.
There was a problem hiding this comment.
thrown is used in most tests because "throw" and "throws" are both java keywords and cannot be used.
There was a problem hiding this comment.
I see. Thank you for clarification.
| @Rule public final transient ExpectedException thrown = ExpectedException.none(); | ||
|
|
||
| @Test | ||
| public void testElementCountConstruction() { |
There was a problem hiding this comment.
Split this test into metric equality test and hashCode equality tests.
Same for user counter version.
Non-equality test missing.
There was a problem hiding this comment.
As discussed, added tests for not equals.
| String urn = SimpleMonitoringInfoBuilder.ELEMENT_COUNT_URN; | ||
| MonitoringInfoMetricName name = MonitoringInfoMetricName.named(urn, labels); | ||
|
|
||
| // Should not fail even though there is no metrics container. |
|
|
||
| @Rule public ExpectedException expectedException = ExpectedException.none(); | ||
|
|
||
| @Before |
There was a problem hiding this comment.
This can be removed. You do not have any mocks defined as class members.
| } | ||
|
|
||
| @Test | ||
| public void multipleConsumersSamePCollection() throws Exception { |
There was a problem hiding this comment.
testCounterIncrementsOnlyOnceWhenMultipleConsumersOfSamePCollectionReadSameElement
There was a problem hiding this comment.
this is too long, I'll just add a comment for this one instead
| } | ||
|
|
||
| @Test | ||
| public void throwsRuntimeExceptionIfRegisteredLate() throws Exception { |
There was a problem hiding this comment.
throwsOnRegisteringPCollectionAfterMultiplexingConsumerWasInitialized
| consumers.register(pCollectionA, consumerA1); | ||
| consumers.getMultiplexingConsumer(pCollectionA); | ||
|
|
||
| // Should throw an exception now if we try to register another after calling |
| } | ||
|
|
||
| // Check that the underlying consumers are each invoked per element | ||
| // and the multiplexing one is invoked only once. |
There was a problem hiding this comment.
Which of the checks validates that multiplexing consumer is invoked only once? I'm misunderstanding the code I guess.
| @@ -68,31 +68,31 @@ func init() { | |||
| runtime.RegisterType(reflect.TypeOf((*meanAccum)(nil)).Elem()) | |||
There was a problem hiding this comment.
I believe this file should not be present in change.
It seem to be formatting change only though.
|
I will resolve these conflicts after PR/7482 is merged |
ajamato
force-pushed
the
element_count_dec14
branch
5 times, most recently
from
8f0ca0b
to
86f796c
Compare
|
Run Python PreCommit |
1 similar comment
|
Run Python PreCommit |
|
Run Java PreCommit |
1 similar comment
|
Run Java PreCommit |
|
Run Python PreCommit") |
ajamato
force-pushed
the
element_count_dec14
branch
from
86f796c
to
9d8fcad
Compare
|
Run Java PreCommit |
…t counters to the java SDK
ajamato
force-pushed
the
element_count_dec14
branch
from
9d8fcad
to
f41b2ef
Compare
|
Run Java PreCommit |
3 similar comments
|
Run Java PreCommit |
|
Run Java PreCommit |
|
Run Java PreCommit |
| @@ -68,6 +68,7 @@ public SpecMonitoringInfoValidator() { | |||
| monitoringInfo.getUrn(), spec.getTypeUrn(), monitoringInfo.getType())); | |||
| } | |||
|
|
|||
| // TODO(ajamato): Tighten this restriction to use set equality, to catch unused | |||
There was a problem hiding this comment.
Just noting: I recently had a local change to keep this a bit relaxed, because I was testing with user metrics that were getting a PTRANSFORM label. I'd weakened the condition below to:
!requiredLabels.ieEmpty && !monitoringInfo.getLabelsMap().keySet().equals(requiredLabels)where previously it required strict equality in all cases.
Seemingly this was relaxed to the containsAll below before I got anywhere with my change, and it's possible I was in some other invalid state to have had user metrics with a PTRANSFORM label that were failing the strict test (where requiredLabels was empty).
Just wanted to leave a breadcrumb here about that since we'll probably come back to it soon..
…t counters to the java SDK
Please add a meaningful description for your change here
Follow this checklist to help us incorporate your contribution quickly and easily:
[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.It will help us expedite review of your Pull Request if you tag someone (e.g.
@username) to look at it.Post-Commit Tests Status (on master branch)