[FLINK-17802][kafka] Set offset commit only if group id is configured…

… for new Kafka Table source

This closes apache#12252

flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -1119,7 +1119,7 @@ LinkedMap getPendingOffsetsToCommit() {
  }
 
  @VisibleForTesting
- boolean getEnableCommitOnCheckpoints() {
+ public boolean getEnableCommitOnCheckpoints() {
  return enableCommitOnCheckpoints;
  }
 

flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSourceBase.java

@@ -209,6 +209,7 @@ protected FlinkKafkaConsumerBase<RowData> getKafkaConsumer(
  kafkaConsumer.setStartFromTimestamp(startupTimestampMillis);
  break;
  }
+ kafkaConsumer.setCommitOffsetsOnCheckpoints(properties.getProperty("group.id") != null);
  return kafkaConsumer;
  }
 }

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java

@@ -170,24 +170,23 @@ public void testTableSource() {
  final StreamExecutionEnvironmentMock mock = new StreamExecutionEnvironmentMock();
  actualKafkaSource.getDataStream(mock);
  assertTrue(getExpectedFlinkKafkaConsumer().isAssignableFrom(mock.sourceFunction.getClass()));
+ // Test commitOnCheckpoints flag should be true when set consumer group.
  assertTrue(((FlinkKafkaConsumerBase) mock.sourceFunction).getEnableCommitOnCheckpoints());
+ }
 
- Properties propsWithoutGroupId = new Properties();
- propsWithoutGroupId.setProperty("bootstrap.servers", "dummy");
-
- final KafkaTableSourceBase sourceWithoutGroupId = getExpectedKafkaTableSource(
- schema,
- Optional.of(PROC_TIME),
- rowtimeAttributeDescriptors,
- fieldMapping,
- TOPIC,
- propsWithoutGroupId,
- deserializationSchema,
- StartupMode.LATEST,
- new HashMap<>(),
- 0L);
-
- sourceWithoutGroupId.getDataStream(mock);
+ @Test
+ public void testTableSourceCommitOnCheckpointsDisabled() {
+ Map<String, String> propertiesMap = new HashMap<>();
+ createKafkaSourceProperties().forEach((k, v) -> {
+ if (!k.equals("connector.properties.group.id")) {
+ propertiesMap.put(k, v);
+ }
+ });
+ final TableSource<?> tableSource = TableFactoryService.find(StreamTableSourceFactory.class, propertiesMap)
+ .createStreamTableSource(propertiesMap);
+ final StreamExecutionEnvironmentMock mock = new StreamExecutionEnvironmentMock();
+ // Test commitOnCheckpoints flag should be false when do not set consumer group.
+ ((KafkaTableSourceBase) tableSource).getDataStream(mock);
  assertTrue(mock.sourceFunction instanceof FlinkKafkaConsumerBase);
  assertFalse(((FlinkKafkaConsumerBase) mock.sourceFunction).getEnableCommitOnCheckpoints());
  }

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTestBase.java

@@ -23,6 +23,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
 import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
@@ -61,7 +62,9 @@
 import static org.apache.flink.util.CoreMatchers.containsCause;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Abstract test base for {@link KafkaDynamicTableFactoryBase}.
@@ -153,6 +156,34 @@ public void testTableSource() {
  final SourceFunctionProvider sourceFunctionProvider = (SourceFunctionProvider) provider;
  final SourceFunction<RowData> sourceFunction = sourceFunctionProvider.createSourceFunction();
  assertThat(sourceFunction, instanceOf(getExpectedConsumerClass()));
+ // Test commitOnCheckpoints flag should be true when set consumer group
+ assertTrue(((FlinkKafkaConsumerBase) sourceFunction).getEnableCommitOnCheckpoints());
+ }
+
+ @Test
+ public void testTableSourceCommitOnCheckpointsDisabled() {
+ //Construct table source using options and table source factory
+ ObjectIdentifier objectIdentifier = ObjectIdentifier.of(
+ "default",
+ "default",
+ "scanTable");
+ Map<String, String> tableOptions = getFullSourceOptions();
+ tableOptions.remove("properties.group.id");
+ CatalogTable catalogTable = createKafkaSourceCatalogTable(tableOptions);
+ final DynamicTableSource tableSource = FactoryUtil.createTableSource(null,
+ objectIdentifier,
+ catalogTable,
+ new Configuration(),
+ Thread.currentThread().getContextClassLoader());
+
+ // Test commitOnCheckpoints flag should be false when do not set consumer group.
+ assertThat(tableSource, instanceOf(KafkaDynamicSourceBase.class));
+ ScanTableSource.ScanRuntimeProvider providerWithoutGroupId = ((KafkaDynamicSourceBase) tableSource)
+ .getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
+ assertThat(providerWithoutGroupId, instanceOf(SourceFunctionProvider.class));
+ final SourceFunctionProvider functionProviderWithoutGroupId = (SourceFunctionProvider) providerWithoutGroupId;
+ final SourceFunction<RowData> function = functionProviderWithoutGroupId.createSourceFunction();
+ assertFalse(((FlinkKafkaConsumerBase) function).getEnableCommitOnCheckpoints());
  }
 
  @Test