[GitHub] [bahir-flink] 1p4pk commented on pull request #151: [BAHIR-308] Bump flink to 1.15

[GitBox <gi...@apache.org>]

1p4pk commented on PR #151:
URL: https://github.com/apache/bahir-flink/pull/151#issuecomment-1230209303

   Hi  @eskabetxe,
   
   sorry for the delay, but yes I spent some time on it and found several things. Also talked to Arvid Heise about it. In general, your version bump leads to the introduction of Soruce/Sink v2 but this connector was built using v1. There are adapters that should guarantee the same behaviour, unfortunately I think is slightly does not.
   
   However, I found to things:
   in the InfluxDBWriter::write, should actually look like that:
   ``` java
   public void write(final IN in, final Context context) throws IOException {
           if (this.elements.size() == this.bufferSize) {
               LOG.debug("Buffer size reached preparing to write the elements.");
               this.writeCurrentElements();
               this.elements.clear();
           }
   
           LOG.trace("Adding elements to buffer. Buffer size: {}", this.elements.size());
           this.elements.add(this.schemaSerializer.serialize(in, context));
           if (context.timestamp() != null) {
               this.lastTimestamp = Math.max(this.lastTimestamp, context.timestamp());
           }
       }
   ```
   
   In the InfluxDBSinkIntegrationTestCase::testSinkDataToInfluxDB, the query should look like that:
   ``` java
   env.addSource(new FiniteTestSource<>(SOURCE_DATA), BasicTypeInfo.LONG_TYPE_INFO)
           .assignTimestampsAndWatermarks(WatermarkStrategy.<Long>noWatermarks()
               .withTimestampAssigner((event, timestamp) -> System.nanoTime()))
           .sinkTo(influxDBSink);
   ```
   
   Of course also add the correct dependencies.
   Besides, I still have the problem that the commits are not generated, which is due to a slight  behviour change by the adapter. Unfortunately, I do not have so much time atm, as I am finishing my master thesis. I can do a drill down a bit later, but not atm. Hope that helped at least a little bit.
   


-- 
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: reviews-unsubscribe@bahir.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org