Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6323#discussion_r202269870
--- Diff:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
---
@@ -134,34 +190,60 @@ public String getProctimeAttribute() {
return rowtimeAttributeDescriptors;
}
+ @Override
+ public Map<String, String> getFieldMapping() {
+ return fieldMapping;
+ }
+
@Override
public String explainSource() {
return TableConnectorUtil.generateRuntimeName(this.getClass(),
schema.getColumnNames());
}
+ /**
+ * Returns the properties for the Kafka consumer.
+ *
+ * @return properties for the Kafka consumer.
+ */
+ public Properties getProperties() {
+ return properties;
+ }
+
+ /**
+ * Returns the deserialization schema.
+ *
+ * @return The deserialization schema
+ */
+ public DeserializationSchema<Row> getDeserializationSchema(){
+ return deserializationSchema;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (!(o instanceof KafkaTableSource)) {
+ // TODO force classes to be equal once we drop support for
format-specific table sources
+ // if (o == null || getClass() != o.getClass()) {
+ if (o == null || !(o instanceof KafkaTableSource)) {
return false;
}
- KafkaTableSource that = (KafkaTableSource) o;
+ final KafkaTableSource that = (KafkaTableSource) o;
return Objects.equals(schema, that.schema) &&
- Objects.equals(topic, that.topic) &&
- Objects.equals(properties, that.properties) &&
- Objects.equals(returnType, that.returnType) &&
Objects.equals(proctimeAttribute,
that.proctimeAttribute) &&
Objects.equals(rowtimeAttributeDescriptors,
that.rowtimeAttributeDescriptors) &&
+ Objects.equals(fieldMapping, that.fieldMapping) &&
+ Objects.equals(topic, that.topic) &&
+ Objects.equals(properties, that.properties) &&
+ Objects.equals(deserializationSchema,
that.deserializationSchema) &&
startupMode == that.startupMode &&
Objects.equals(specificStartupOffsets,
that.specificStartupOffsets);
}
@Override
public int hashCode() {
- return Objects.hash(schema, topic, properties, returnType,
- proctimeAttribute, rowtimeAttributeDescriptors,
startupMode, specificStartupOffsets);
+ return Objects.hash(schema, proctimeAttribute,
rowtimeAttributeDescriptors, fieldMapping,
--- End diff --
format one entry per line
---
