You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by Bekreth <gi...@git.apache.org> on 2018/02/20 20:39:07 UTC
[GitHub] flink pull request #5538: [FLINK-8655] [DataSink] Added default keyspace to ...
GitHub user Bekreth opened a pull request:
https://github.com/apache/flink/pull/5538
[FLINK-8655] [DataSink] Added default keyspace to CassandraPojoSink
## What is the purpose of the change
Initial Issue : [https://issues.apache.org/jira/browse/FLINK-8655]
It is not uncommon for users to have a single Cassandra instance for all of their test environments, differentiating what each environment uses based on keyspace. This PR adds functionality for the CassandraPojoSink to have the keyspace defined at runtime to allow for more flexible configurations.
## Brief change log
- Added functionality for defining default keyspace at runtime.
## Verifying this change
The changes have been verified manually. Unit tests will be included following preliminary discussion on the functionality.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/Bekreth/flink cassandraDefaultKeyspaceSink
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5538.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5538
----
commit bbf52274ba46da005894c1f7c4be4860cd619280
Author: chugh13 <ch...@...>
Date: 2018-02-20T20:02:11Z
Added method to specify the keyspace that POJOs should be written to by default.
commit 0e014aefbb7cccf0e1d8caed50f2a9dc88962d4f
Author: chugh13 <ch...@...>
Date: 2018-02-20T20:37:17Z
Added Javadocs and log messages to the defaultKeyspace feature
----
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by Bekreth <gi...@git.apache.org>.
Github user Bekreth commented on the issue:
https://github.com/apache/flink/pull/5538
Previously, I was trying to leverage the current annotation parsers.
It would be possible to pass in a keyspace more directly into AnotationParser in the shaded Datastax component, but I'm against this as it requires maintaining a version Datastax that differs from the open source version.
It could be possible to just replace the entire annotation with reflections instead of just editing 1 property, but that still requires some futzing in reflections that I think would be best to avoid.
Another option could be to extend the necessary Datastax classes within the CassandraPojoSink to facilitate the alterations necessary to pass in keyspace dynamically. I'm thinking this is the most favorable option.
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/5538
I see the need for the feature, but I am skeptical concerning the implementation.
This PR reflectively modifies the contents of a String. This is prone to cause problems, for multiple reasons. Strings cache hash codes, and strings themselves are interned and shared.
Is there no other way to pass a default namespace to the mapping?
---
[GitHub] flink pull request #5538: [FLINK-8655] [DataSink] Added default keyspace to ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5538
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by Bekreth <gi...@git.apache.org>.
Github user Bekreth commented on the issue:
https://github.com/apache/flink/pull/5538
I've been looking into this more in depth and I'm uncertain there is going to be a cleaner solution. The DataStax annotation parser will fail during runtime if it can't determine the keyspace either from the Table annotation, or from the connection. The core Cassandra functionality is built on a stateless connection though, so embedding the keyspace into there would be a bad idea.
In order to make this work with the current Datastax implementations, it'd be necessary to extend both MappingManager and AnnotationParser from Datastax as well as update the CassandraSinkBase to use these extension.
Should I try to make a swing at these changes to see what it entail?
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5538
We would have to pass the keyspace via the constructor as the `Configuration` approach doesn't work for streaming.
Generally speaking it isn't a problem to set the keyspace when creating the connection. But I would like to know what happens if a POJO comes along that explicitly sets the keyspace; is ignored, respected or will it cause an exception?
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by ctamisier <gi...@git.apache.org>.
Github user ctamisier commented on the issue:
https://github.com/apache/flink/pull/5538
This is also something I would need and I would like to see with you for an alternative fix.
What about using the `Configuration` that is provided in `RichFunction.open(Configuration parameters)` for the `CassandraSinkBase.open(Configuration configuration) {...}` implementation ?
I saw in [old docs](https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/best_practices.html#passing-parameters-as-a-configuration-object-to-single-functions) that `Configuration` can be used in the `open(...)` method but today (1.4+) it might not be a good practice anymore.
What about adding `keyspace` attribute in `CassandraPojoSink` and `CassandraSinkBuilder` (throwing exception when not using a `CassandraPojoSinkBuilder` for the moment).
And create a new `Configuration()` with this `keyspace` in `CassandraPojoSink`.
And finally do a `cluster.connect(keyspace);`
I've done this [here](https://github.com/apache/flink/compare/master...ctamisier:pojo-keyspace) if you can have a look.
I've updated `CassandraConnectorITCase` with a new test.
I would like to run `CassandraPojoSinkExample.main()` to cover the `CassandraSink.addSink()` mechanism, but it doesn't work for me (even on flink master branch).
Can this be a candidate for a PR, I am new to flink so it might break the flink good practice principles...
Let me know!
---
[GitHub] flink issue #5538: [FLINK-8655] [DataSink] Added default keyspace to Cassand...
Posted by Bekreth <gi...@git.apache.org>.
Github user Bekreth commented on the issue:
https://github.com/apache/flink/pull/5538
This looks like it'd work out well! When I made this PR, I was aware that the Cassandra connection happened irrespective of the keyspace and I tried to maintain that in order to minimize the change made. If the Flink team will to allow stateful connections (`CassandraSinkBase:L88`), then I'd much prefer to see them use your solution.
---