Skip to content

Commit 96964f5

Browse files
rajuGTrajuGT
rajuGT
and
rajuGT
authored
feat/Influx sink - support multiple retention policies (#70)
* feat/influx-sink - Support multiple retention policies Context: Some of the measurements/series you're interested in live data, old data is not so useful/interesting OR Say you've high frequency influx points and you've summarry/aggregated points. You would like to keep summary points for over weeks/months whereas granular/high frequency data can't be kept for that duration For the above mentioned cases - one can utilize the retention policies where; same db, same measurement, but different retention policy. Details: Add a new configuration SINK_INFLUX_RETENTION_POLICY_LIST_KEY which is similar to SINK_INFLUX_MEASUREMENTS_LIST_KEY and the usage similar how we used multiple measurements. In custom jobs we would like to configure the sink like ``` measurements = String.split(configuration.getString(SINK_INFLUX_MEASUREMENTS_LIST_KEY), ',') measurementName = measurements[0] retention_policies = String.split(configuration.getString(SINK_INFLUX_RETENTION_POLICY_LIST_KEY), ',') retentionPolicy = retention_policies[1] getSink(measurement_name, retentionPolicy, .....and other args) ``` * refactor - overrides are packaged in an object instead of increasing parameters of the methods * fix/ checkstyle issue * fix/ make InfluxSinkOverrides serializable --------- Co-authored-by: rajuGT <raju.gt@gojek.com>
1 parent e0f2f38 commit 96964f5

11 files changed

Lines changed: 226 additions & 53 deletions

File tree

‎dagger-core/src/main/java/com/gotocompany/dagger/core/DaggerSqlJobBuilder.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import com.gotocompany.dagger.core.processors.types.PostProcessor;
1717
import com.gotocompany.dagger.core.processors.types.Preprocessor;
1818
import com.gotocompany.dagger.core.sink.SinkOrchestrator;
19+
import com.gotocompany.dagger.core.sink.influx.InfluxSinkOverrides;
1920
import com.gotocompany.dagger.core.source.StreamsFactory;
2021
import com.gotocompany.dagger.core.utils.Constants;
2122
import com.gotocompany.dagger.core.processors.telemetry.processor.MetricsTelemetryExporter;
@@ -230,6 +231,6 @@ private StreamInfo addPreProcessor(StreamInfo streamInfo, String tableName) {
230231
private void addSink(StreamInfo streamInfo) {
231232
SinkOrchestrator sinkOrchestrator = new SinkOrchestrator(telemetryExporter);
232233
sinkOrchestrator.addSubscriber(telemetryExporter);
233-
streamInfo.getDataStream().sinkTo(sinkOrchestrator.getSink(configuration, streamInfo.getColumnNames(), stencilClientOrchestrator, daggerStatsDReporter));
234+
streamInfo.getDataStream().sinkTo(sinkOrchestrator.getSink(configuration, streamInfo.getColumnNames(), stencilClientOrchestrator, daggerStatsDReporter, InfluxSinkOverrides.none()));
234235
}
235236
}

‎dagger-core/src/main/java/com/gotocompany/dagger/core/ExampleStreamApiJobBuilder.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import com.gotocompany.dagger.core.processors.telemetry.processor.MetricsTelemetryExporter;
1313
import com.gotocompany.dagger.core.processors.types.Preprocessor;
1414
import com.gotocompany.dagger.core.sink.SinkOrchestrator;
15+
import com.gotocompany.dagger.core.sink.influx.InfluxSinkOverrides;
1516
import com.gotocompany.dagger.core.source.StreamsFactory;
1617
import com.gotocompany.dagger.core.utils.Constants;
1718
import org.apache.flink.streaming.api.CheckpointingMode;
@@ -137,7 +138,7 @@ public JobBuilder registerOutputStream() {
137138
.keyBy(r -> 0)
138139
.max("someField");
139140

140-
outputStream.sinkTo(sinkOrchestrator.getSink(configuration, new String[]{"uniq_users"}, stencilClientOrchestrator, daggerStatsDReporter));
141+
outputStream.sinkTo(sinkOrchestrator.getSink(configuration, new String[]{"uniq_users"}, stencilClientOrchestrator, daggerStatsDReporter, InfluxSinkOverrides.none()));
141142
return this;
142143
}
143144

‎dagger-core/src/main/java/com/gotocompany/dagger/core/sink/SinkOrchestrator.java‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import com.gotocompany.dagger.core.sink.influx.ErrorHandler;
99
import com.gotocompany.dagger.core.sink.influx.InfluxDBFactoryWrapper;
1010
import com.gotocompany.dagger.core.sink.influx.InfluxDBSink;
11+
import com.gotocompany.dagger.core.sink.influx.InfluxSinkOverrides;
1112
import com.gotocompany.dagger.core.utils.KafkaConfigUtil;
1213
import com.gotocompany.dagger.core.utils.Constants;
1314
import org.apache.flink.api.connector.sink.Sink;
@@ -47,13 +48,15 @@ public SinkOrchestrator(MetricsTelemetryExporter telemetryExporter) {
4748
/**
4849
* Gets sink.
4950
*
51+
* @param configuration the configuration
52+
* @param columnNames the column names
53+
* @param stencilClientOrchestrator the stencil client orchestrator
54+
* @param daggerStatsDReporter the StatsD reporter
55+
* @param influxSinkOverrides Influx sink overrides; pass {@link InfluxSinkOverrides#none()} when no influxSinkOverrides are needed
5056
* @return the sink
51-
* @configuration configuration the configuration
52-
* @columnNames columnNames the column names
53-
* @StencilClientOrchestrator stencilClientOrchestrator the stencil client orchestrator
5457
*/
5558
public Sink getSink(Configuration configuration, String[] columnNames, StencilClientOrchestrator stencilClientOrchestrator,
56-
DaggerStatsDReporter daggerStatsDReporter, String influxMeasurementOverrideName) {
59+
DaggerStatsDReporter daggerStatsDReporter, InfluxSinkOverrides influxSinkOverrides) {
5760
String sinkType = configuration.getString("SINK_TYPE", "influx");
5861
addMetric(TelemetryTypes.SINK_TYPE.getValue(), sinkType);
5962
Sink sink;
@@ -86,18 +89,12 @@ public Sink getSink(Configuration configuration, String[] columnNames, StencilCl
8689
.build();
8790
break;
8891
default:
89-
sink = new InfluxDBSink(new InfluxDBFactoryWrapper(), configuration, columnNames, new ErrorHandler(), influxMeasurementOverrideName);
92+
sink = new InfluxDBSink(new InfluxDBFactoryWrapper(), configuration, columnNames, new ErrorHandler(), influxSinkOverrides);
9093
}
9194
notifySubscriber();
9295
return sink;
9396
}
9497

95-
public Sink getSink(Configuration configuration, String[] columnNames, StencilClientOrchestrator stencilClientOrchestrator,
96-
DaggerStatsDReporter daggerStatsDReporter) {
97-
String influxMeasurementOverrideName = null;
98-
return getSink(configuration, columnNames, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
99-
}
100-
10198
private void reportTelemetry(KafkaSerializerBuilder kafkaSchemaBuilder) {
10299
TelemetryPublisher pub = (TelemetryPublisher) kafkaSchemaBuilder;
103100
pub.addSubscriber(telemetryExporter);

‎dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBSink.java‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,15 +25,15 @@ public class InfluxDBSink implements Sink<Row, Void, Void, Void> {
2525
private String[] columnNames;
2626
private ErrorHandler errorHandler;
2727
private ErrorReporter errorReporter;
28-
private final String influxMeasurementOverrideName;
28+
private final InfluxSinkOverrides overrides;
2929

3030
public InfluxDBSink(InfluxDBFactoryWrapper influxDBFactory, Configuration configuration, String[] columnNames,
31-
ErrorHandler errorHandler, String influxMeasurementOverrideName) {
31+
ErrorHandler errorHandler, InfluxSinkOverrides overrides) {
3232
this.influxDBFactory = influxDBFactory;
3333
this.configuration = configuration;
3434
this.columnNames = columnNames;
3535
this.errorHandler = errorHandler;
36-
this.influxMeasurementOverrideName = influxMeasurementOverrideName;
36+
this.overrides = overrides == null ? InfluxSinkOverrides.none() : overrides;
3737
}
3838

3939
@Override
@@ -49,7 +49,7 @@ public SinkWriter<Row, Void, Void> createWriter(InitContext context, List<Void>
4949
errorReporter = ErrorReporterFactory.getErrorReporter(context.metricGroup(), configuration);
5050
}
5151

52-
InfluxDBWriter influxDBWriter = new InfluxDBWriter(configuration, influxDB, columnNames, errorHandler, errorReporter, influxMeasurementOverrideName);
52+
InfluxDBWriter influxDBWriter = new InfluxDBWriter(configuration, influxDB, columnNames, errorHandler, errorReporter, overrides);
5353
return influxDBWriter;
5454
}
5555

@@ -77,5 +77,4 @@ public Optional<SimpleVersionedSerializer<Void>> getCommittableSerializer() {
7777
public Optional<SimpleVersionedSerializer<Void>> getGlobalCommittableSerializer() {
7878
return Optional.empty();
7979
}
80-
8180
}

‎dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -36,15 +36,16 @@ public class InfluxDBWriter implements SinkWriter<Row, Void, Void> {
3636
private ErrorReporter errorReporter;
3737
private boolean useRowFieldNames;
3838

39+
3940
public InfluxDBWriter(Configuration configuration, InfluxDB influxDB, String[] columnNames, ErrorHandler errorHandler,
40-
ErrorReporter errorReporter, String influxMeasurementOverrideName) {
41+
ErrorReporter errorReporter, InfluxSinkOverrides overrides) {
4142
databaseName = configuration.getString(Constants.SINK_INFLUX_DB_NAME_KEY, Constants.SINK_INFLUX_DB_NAME_DEFAULT);
42-
retentionPolicy = configuration.getString(Constants.SINK_INFLUX_RETENTION_POLICY_KEY, Constants.SINK_INFLUX_RETENTION_POLICY_DEFAULT);
43-
if (Strings.isNullOrEmpty(influxMeasurementOverrideName)) {
44-
measurementName = configuration.getString(Constants.SINK_INFLUX_MEASUREMENT_NAME_KEY, Constants.SINK_INFLUX_MEASUREMENT_NAME_DEFAULT);
45-
} else {
46-
measurementName = influxMeasurementOverrideName;
47-
}
43+
retentionPolicy = overrides.hasRetentionPolicy()
44+
? overrides.getRetentionPolicy()
45+
: configuration.getString(Constants.SINK_INFLUX_RETENTION_POLICY_KEY, Constants.SINK_INFLUX_RETENTION_POLICY_DEFAULT);
46+
measurementName = overrides.hasMeasurementName()
47+
? overrides.getMeasurementName()
48+
: configuration.getString(Constants.SINK_INFLUX_MEASUREMENT_NAME_KEY, Constants.SINK_INFLUX_MEASUREMENT_NAME_DEFAULT);
4849
useRowFieldNames = configuration.getBoolean(Constants.SINK_INFLUX_USING_ROW_FIELD_NAMES_KEY, Constants.SINK_INFLUX_USING_ROW_FIELD_NAMES_DEFAULT);
4950
this.influxDB = influxDB;
5051
this.columnNames = columnNames;

‎dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
package com.gotocompany.dagger.core.sink.influx;
2+
3+
import com.google.common.base.Strings;
4+
5+
import java.io.Serializable;
6+
import java.util.Objects;
7+
8+
/**
9+
* Immutable value object that carries optional overrides for an Influx sink:
10+
* <ul>
11+
* <li>measurement name (overrides {@code SINK_INFLUX_MEASUREMENT_NAME})</li>
12+
* <li>retention policy (overrides {@code SINK_INFLUX_RETENTION_POLICY})</li>
13+
* </ul>
14+
*
15+
* Either field may be {@code null} or blank, which means "fall back to the
16+
* value resolved from configuration". Using this object instead of multiple
17+
* positional {@code String} parameters keeps sink construction readable as
18+
* the number of overrides grows.
19+
*
20+
* <p>Typical usage from a custom job:
21+
* <pre>{@code
22+
* InfluxSinkOverrides overrides = InfluxSinkOverrides.builder()
23+
* .measurementName(measurements[i])
24+
* .retentionPolicy(retentionPolicies[i])
25+
* .build();
26+
* Sink sink = sinkOrchestrator.getSink(configuration, columnNames,
27+
* stencilClientOrchestrator, daggerStatsDReporter, influxOverrides);
28+
* }</pre>
29+
*/
30+
public final class InfluxSinkOverrides implements Serializable {
31+
32+
private static final long serialVersionUID = 1L;
33+
34+
private static final InfluxSinkOverrides NONE = new InfluxSinkOverrides(null, null);
35+
36+
private final String measurementName;
37+
private final String retentionPolicy;
38+
39+
private InfluxSinkOverrides(String measurementName, String retentionPolicy) {
40+
this.measurementName = measurementName;
41+
this.retentionPolicy = retentionPolicy;
42+
}
43+
44+
/** Returns an instance that applies no overrides (configuration values are used). */
45+
public static InfluxSinkOverrides none() {
46+
return NONE;
47+
}
48+
49+
public static InfluxSinkOverrides withMeasurementName(String measurementName) {
50+
return new InfluxSinkOverrides(measurementName, null);
51+
}
52+
53+
public static InfluxSinkOverrides withRetentionPolicy(String retentionPolicy) {
54+
return new InfluxSinkOverrides(null, retentionPolicy);
55+
}
56+
57+
public static InfluxSinkOverrides of(String measurementName, String retentionPolicy) {
58+
return new InfluxSinkOverrides(measurementName, retentionPolicy);
59+
}
60+
61+
public static Builder builder() {
62+
return new Builder();
63+
}
64+
65+
/** May be {@code null} or empty, meaning "use the value from configuration". */
66+
public String getMeasurementName() {
67+
return measurementName;
68+
}
69+
70+
/** May be {@code null} or empty, meaning "use the value from configuration". */
71+
public String getRetentionPolicy() {
72+
return retentionPolicy;
73+
}
74+
75+
public boolean hasMeasurementName() {
76+
return !Strings.isNullOrEmpty(measurementName);
77+
}
78+
79+
public boolean hasRetentionPolicy() {
80+
return !Strings.isNullOrEmpty(retentionPolicy);
81+
}
82+
83+
@Override
84+
public boolean equals(Object o) {
85+
if (this == o) {
86+
return true;
87+
}
88+
if (!(o instanceof InfluxSinkOverrides)) {
89+
return false;
90+
}
91+
InfluxSinkOverrides that = (InfluxSinkOverrides) o;
92+
return Objects.equals(measurementName, that.measurementName)
93+
&& Objects.equals(retentionPolicy, that.retentionPolicy);
94+
}
95+
96+
@Override
97+
public int hashCode() {
98+
return Objects.hash(measurementName, retentionPolicy);
99+
}
100+
101+
@Override
102+
public String toString() {
103+
return "InfluxSinkOverrides{measurementName='" + measurementName
104+
+ "', retentionPolicy='" + retentionPolicy + "'}";
105+
}
106+
107+
/** Fluent builder for {@link InfluxSinkOverrides}. */
108+
public static final class Builder {
109+
private String measurementName;
110+
private String retentionPolicy;
111+
112+
private Builder() {
113+
}
114+
115+
public Builder measurementName(String name) {
116+
this.measurementName = name;
117+
return this;
118+
}
119+
120+
public Builder retentionPolicy(String policy) {
121+
this.retentionPolicy = policy;
122+
return this;
123+
}
124+
125+
public InfluxSinkOverrides build() {
126+
if (measurementName == null && retentionPolicy == null) {
127+
return NONE;
128+
}
129+
return new InfluxSinkOverrides(measurementName, retentionPolicy);
130+
}
131+
}
132+
}

‎dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,12 @@ public class Constants {
163163
// If measurement names need to change, they can now be updated through configuration without changing the code.
164164
public static final String SINK_INFLUX_MEASUREMENTS_LIST_KEY = "SINK_INFLUX_MEASUREMENTS_LIST";
165165

166+
// A custom job can use this configuration to get all Influx retention policies as a list
167+
// (positionally aligned with SINK_INFLUX_MEASUREMENTS_LIST_KEY) and configure them in the
168+
// job builder pipeline accordingly. This allows retention policies to be updated through
169+
// configuration without changing the code.
170+
public static final String SINK_INFLUX_RETENTION_POLICY_LIST_KEY = "SINK_INFLUX_RETENTION_POLICY_LIST";
171+
166172
public static final String SINK_INFLUX_URL_KEY = "SINK_INFLUX_URL";
167173
public static final String SINK_INFLUX_URL_DEFAULT = "";
168174
public static final String SINK_INFLUX_USERNAME_KEY = "SINK_INFLUX_USERNAME";

‎dagger-core/src/test/java/com/gotocompany/dagger/core/sink/SinkOrchestratorTest.java‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import com.gotocompany.dagger.core.processors.telemetry.processor.MetricsTelemetryExporter;
77
import com.gotocompany.dagger.core.sink.bigquery.BigQuerySink;
88
import com.gotocompany.dagger.core.sink.influx.InfluxDBSink;
9+
import com.gotocompany.dagger.core.sink.influx.InfluxSinkOverrides;
910
import com.gotocompany.dagger.core.sink.log.LogSink;
1011
import com.gotocompany.dagger.core.utils.Constants;
1112
import org.apache.flink.api.connector.sink.Sink;
@@ -25,7 +26,7 @@
2526
import static org.mockito.MockitoAnnotations.initMocks;
2627

2728
public class SinkOrchestratorTest {
28-
private final String influxMeasurementOverrideName = "";
29+
private final InfluxSinkOverrides influxSinkOverrides = InfluxSinkOverrides.none();
2930

3031
private static final String SINK_KAFKA_PRODUCER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS = "SINK_KAFKA_PRODUCER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS";
3132
private static final String SASL_LOGIN_CALLBACK_HANDLER_CLASS_VALUE = "com.gotocompany.dagger.core.utils.SinkKafkaConfigUtil";
@@ -52,23 +53,23 @@ public void setup() {
5253
@Test
5354
public void shouldGiveInfluxSinkWhenConfiguredToUseInflux() throws Exception {
5455
when(configuration.getString(eq("SINK_TYPE"), anyString())).thenReturn("influx");
55-
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
56+
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxSinkOverrides);
5657

5758
assertThat(sinkFunction, instanceOf(InfluxDBSink.class));
5859
}
5960

6061
@Test
6162
public void shouldGiveLogSinkWhenConfiguredToUseLog() throws Exception {
6263
when(configuration.getString(eq("SINK_TYPE"), anyString())).thenReturn("log");
63-
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
64+
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxSinkOverrides);
6465

6566
assertThat(sinkFunction, instanceOf(LogSink.class));
6667
}
6768

6869
@Test
6970
public void shouldGiveInfluxWhenConfiguredToUseNothing() throws Exception {
7071
when(configuration.getString(eq("SINK_TYPE"), anyString())).thenReturn("");
71-
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
72+
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxSinkOverrides);
7273

7374
assertThat(sinkFunction, instanceOf(InfluxDBSink.class));
7475
}
@@ -108,7 +109,7 @@ public void shouldReturnSinkMetrics() {
108109

109110
when(configuration.getString(eq("SINK_TYPE"), anyString())).thenReturn("influx");
110111

111-
sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
112+
sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxSinkOverrides);
112113
assertEquals(expectedMetrics, sinkOrchestrator.getTelemetry());
113114
}
114115

@@ -117,7 +118,7 @@ public void shouldReturnBigQuerySink() {
117118
when(configuration.getString(eq("SINK_TYPE"), anyString())).thenReturn("bigquery");
118119
when(configuration.getString("SINK_CONNECTOR_SCHEMA_PROTO_MESSAGE_CLASS", "")).thenReturn("some.class");
119120
when(configuration.getParam()).thenReturn(ParameterTool.fromMap(Collections.emptyMap()));
120-
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxMeasurementOverrideName);
121+
Sink sinkFunction = sinkOrchestrator.getSink(configuration, new String[]{}, stencilClientOrchestrator, daggerStatsDReporter, influxSinkOverrides);
121122
assertThat(sinkFunction, instanceOf(BigQuerySink.class));
122123
}
123124
}

‎dagger-core/src/test/java/com/gotocompany/dagger/core/sink/influx/InfluxDBSinkTest.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
import static org.mockito.MockitoAnnotations.initMocks;
3030

3131
public class InfluxDBSinkTest {
32-
private final String influxMeasurementOverrideName = "";
32+
private final InfluxSinkOverrides influxSinkOverrides = InfluxSinkOverrides.none();
3333
private static final int SINK_INFLUX_BATCH_SIZE = 100;
3434
private static final int INFLUX_FLUSH_DURATION = 1000;
3535

@@ -74,7 +74,7 @@ public void setUp() throws Exception {
7474

7575
@Test
7676
public void shouldCallInfluxDbFactoryWhileCreatingWriter() throws Exception {
77-
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxMeasurementOverrideName);
77+
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxSinkOverrides);
7878
List<Void> state = new ArrayList<>();
7979
influxDBSink.createWriter(context, state);
8080

@@ -83,7 +83,7 @@ public void shouldCallInfluxDbFactoryWhileCreatingWriter() throws Exception {
8383

8484
@Test
8585
public void shouldCreateInfluxWriter() throws IOException {
86-
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxMeasurementOverrideName);
86+
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxSinkOverrides);
8787
List<Void> state = new ArrayList<>();
8888
SinkWriter<Row, Void, Void> writer = influxDBSink.createWriter(context, state);
8989

@@ -92,7 +92,7 @@ public void shouldCreateInfluxWriter() throws IOException {
9292

9393
@Test
9494
public void shouldCallBatchModeOnInfluxWhenBatchSettingsExist() throws Exception {
95-
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxMeasurementOverrideName);
95+
InfluxDBSink influxDBSink = new InfluxDBSink(influxDBFactory, configuration, new String[]{}, errorHandler, influxSinkOverrides);
9696
List<Void> state = new ArrayList<>();
9797
influxDBSink.createWriter(context, state);
9898
verify(influxDb).enableBatch(eq(SINK_INFLUX_BATCH_SIZE), eq(INFLUX_FLUSH_DURATION), eq(TimeUnit.MILLISECONDS), any(ThreadFactory.class), any(BiConsumer.class));

0 commit comments

Comments
 (0)