+1 :)
On 08/05/17 23:52, Matthias J. Sax wrote:
Hi,
I was reading the updated KIP and I am wondering, if we should do the
design a little different.
Instead of distinguishing between a RichFunction and non-RichFunction at
runtime level, we would use RichFunctions all the time. Thus, on the DSL
entry level, if a user provides a non-RichFunction, we wrap it by a
RichFunction that is fully implemented by Streams. This RichFunction
would just forward the call omitting the key:
Just to sketch the idea (incomplete code snippet):
public StreamsRichValueMapper implements RichValueMapper() {
private ValueMapper userProvidedMapper; // set by constructor
public VR apply(final K key, final V1 value1, final V2 value2) {
return userProvidedMapper(value1, value2);
}
}
From a performance point of view, I am not sure if the
"if(isRichFunction)" including casts etc would have more overhead than
this approach (we would do more nested method call for non-RichFunction
which should be more common than RichFunctions).
This approach should not effect lambdas (or do I miss something?) and
might be cleaner, as we could have one more top level interface
`RichFunction` with methods `init()` and `close()` and also interfaces
for `RichValueMapper` etc. (thus, no abstract classes required).
Any thoughts?
-Matthias
On 5/6/17 5:29 PM, Jeyhun Karimov wrote:
Hi,
Thanks for comments. I extended PR and KIP to include rich functions. I
will still have to evaluate the cost of deep copying of keys.
Cheers,
Jeyhun
On Fri, May 5, 2017 at 8:02 PM Mathieu Fenniak <mathieu.fenn...@replicon.com>
wrote:
Hey Matthias,
My opinion would be that documenting the immutability of the key is the
best approach available. Better than requiring the key to be serializable
(as with Jeyhun's last pass at the PR), no performance risk.
It'd be different if Java had immutable type constraints of some kind. :-)
Mathieu
On Fri, May 5, 2017 at 11:31 AM, Matthias J. Sax <matth...@confluent.io>
wrote:
Agreed about RichFunction. If we follow this path, it should cover
all(?) DSL interfaces.
About guarding the key -- I am still not sure what to do about it...
Maybe it might be enough to document it (and name the key parameter like
`readOnlyKey` to make is very clear). Ultimately, I would prefer to
guard against any modification, but I have no good idea how to do this.
Not sure what others think about the risk of corrupted partitioning
(what would be a user error and we could say, well, bad luck, you got a
bug in your code, that's not our fault), vs deep copy with a potential
performance hit (that we can't quantity atm without any performance
test).
We do have a performance system test. Maybe it's worth for you to apply
the deep copy strategy and run the test. It's very basic performance
test only, but might give some insight. If you want to do this, look
into folder "tests" for general test setup, and into
"tests/kafaktests/benchmarks/streams" to find find the perf test.
-Matthias
On 5/5/17 8:55 AM, Jeyhun Karimov wrote:
Hi Matthias,
I think extending KIP to include RichFunctions totally makes sense.
So,
we don't want to guard the keys because it is costly.
If we introduce RichFunctions I think it should not be limited only
the 3
interfaces proposed in KIP but to wide range of interfaces.
Please correct me if I am wrong.
Cheers,
Jeyhun
On Fri, May 5, 2017 at 12:04 AM Matthias J. Sax <matth...@confluent.io
wrote:
One follow up. There was this email on the user list:
http://search-hadoop.com/m/Kafka/uyzND17KhCaBzPSZ1?subj=
Shouldn+t+the+initializer+of+a+stream+aggregate+accept+the+key+
It might make sense so include Initializer, Adder, and Substractor
inferface, too.
And we should double check if there are other interface we might miss
atm.
-Matthias
On 5/4/17 1:31 PM, Matthias J. Sax wrote:
Thanks for updating the KIP.
Deep copying the key will work for sure, but I am actually a little
bit
worried about performance impact... We might want to do some test to
quantify this impact.
Btw: this remind me about the idea of `RichFunction` interface that
would allow users to access record metadata (like timestamp, offset,
partition etc) within DSL. This would be a similar concept. Thus, I
am
wondering, if it would make sense to enlarge the scope of this KIP by
that? WDYT?
-Matthias
On 5/3/17 2:08 AM, Jeyhun Karimov wrote:
Hi Mathieu,
Thanks for feedback. I followed similar approach and updated PR and
KIP
accordingly. I tried to guard the key in Processors sending a copy
of
an
actual key.
Because I am doing deep copy of an object, I think memory can be
bottleneck
in some use-cases.
Cheers,
Jeyhun
On Tue, May 2, 2017 at 5:10 PM Mathieu Fenniak <
mathieu.fenn...@replicon.com>
wrote:
Hi Jeyhun,
This approach would change ValueMapper (...etc) to be classes,
rather
than
interfaces, which is also a backwards incompatible change. An
alternative
approach that would be backwards compatible would be to define new
interfaces, and provide overrides where those interfaces are used.
Unfortunately, making the key parameter as "final" doesn't change
much
about guarding against key change. It only prevents the parameter
variable
from being reassigned. If the key type is a mutable object (eg.
byte[]),
it can still be mutated. (eg. key[0] = 0). But I'm not really sure
there's
much that can be done about that.
Mathieu
On Mon, May 1, 2017 at 5:39 PM, Jeyhun Karimov <
je.kari...@gmail.com
wrote:
Thanks for comments.
The concerns makes sense. Although we can guard for immutable keys
in
current implementation (with few changes), I didn't consider
backward
compatibility.
In this case 2 solutions come to my mind. In both cases, user
accesses
the
key in Object type, as passing extra type parameter will break
backwards-compatibility. So user has to cast to actual key type.
1. Firstly, We can overload apply method with 2 argument (key and
value)
and force key to be *final*. By doing this, I think we can
address
both
backward-compatibility and guarding against key change.
2. Secondly, we can create class KeyAccess like:
public class KeyAccess {
Object key;
public void beforeApply(final Object key) {
this.key = key;
}
public Object getKey() {
return key;
}
}
We can extend *ValueMapper, ValueJoiner* and *ValueTransformer*
from
*KeyAccess*. Inside processor (for example
*KTableMapValuesProcessor*)
before calling *mapper.apply(value)* we can set the *key* by
*mapper.beforeApply(key)*. As a result, user can use *getKey()* to
access
the key inside *apply(value)* method.
Cheers,
Jeyhun
On Mon, May 1, 2017 at 7:24 PM Matthias J. Sax <
matth...@confluent.io
wrote:
Jeyhun,
thanks a lot for the KIP!
I think there are two issues we need to address:
(1) The KIP does not consider backward compatibility. Users did
complain
about this in past releases already, and as the user base grows,
we
should not break backward compatibility in future releases
anymore.
Thus, we should think of a better way to allow key access.
Mathieu's comment goes into the same direction
On the other hand, the number of compile failures that would
need
to
be
fixed from this change is unfortunate. :-)
(2) Another concern is, that there is no guard to prevent user
code
to
modify the key. This might corrupt partitioning if users do alter
the
key (accidentally -- or users are just not aware that they are
not
allowed to modify the provided key object) and thus break the
application. (This was the original motivation to not provide the
key
in
the first place -- it's guards against modification.)
-Matthias
On 5/1/17 6:31 AM, Mathieu Fenniak wrote:
Hi Jeyhun,
I just want to add my voice that, I too, have wished for access
to
the
record key during a mapValues or similar operation.
On the other hand, the number of compile failures that would
need
to
be
fixed from this change is unfortunate. :-) But at least it
would
all
be
a
pretty clear and easy change.
Mathieu
On Mon, May 1, 2017 at 6:55 AM, Jeyhun Karimov <
je.kari...@gmail.com
wrote:
Dear community,
I want to share KIP-149 [1] based on issues KAFKA-4218 [2],
KAFKA-4726
[3],
KAFKA-3745 [4]. The related PR can be found at [5].
I would like to get your comments.
[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-
149%3A+Enabling+key+access+in+ValueTransformer%2C+
ValueMapper%2C+and+ValueJoiner
[2] https://issues.apache.org/jira/browse/KAFKA-4218
[3] https://issues.apache.org/jira/browse/KAFKA-4726
[4] https://issues.apache.org/jira/browse/KAFKA-3745
[5] https://github.com/apache/kafka/pull/2946
Cheers,
Jeyhun
--
-Cheers
Jeyhun
--
-Cheers
Jeyhun
--
-Cheers
Jeyhun
