@Test public void testTableSink() { final TableSchema schema = createTestSchema(); final ElasticsearchUpsertTableSinkBase expectedSink = getExpectedTableSink( false, schema, Collections.singletonList(new Host(HOSTNAME, PORT, SCHEMA)), INDEX, DOC_TYPE, KEY_DELIMITER, KEY_NULL_LITERAL, new JsonRowSerializationSchema(schema.toRowType()), XContentType.JSON, new DummyFailureHandler(), createTestSinkOptions()); final TestTableDescriptor testDesc = new TestTableDescriptor( new Elasticsearch() .version(getElasticsearchVersion()) .host(HOSTNAME, PORT, SCHEMA) .index(INDEX) .documentType(DOC_TYPE) .keyDelimiter(KEY_DELIMITER) .keyNullLiteral(KEY_NULL_LITERAL) .bulkFlushBackoffExponential() .bulkFlushBackoffDelay(123L) .bulkFlushBackoffMaxRetries(3) .bulkFlushInterval(100L) .bulkFlushMaxActions(1000) .bulkFlushMaxSize("1 MB") .failureHandlerCustom(DummyFailureHandler.class) .connectionMaxRetryTimeout(100) .connectionPathPrefix("/myapp")) .withFormat( new Json() .deriveSchema()) .withSchema( new Schema() .field(FIELD_KEY, Types.LONG()) .field(FIELD_FRUIT_NAME, Types.STRING()) .field(FIELD_COUNT, Types.DECIMAL()) .field(FIELD_TS, Types.SQL_TIMESTAMP())) .inUpsertMode(); final Map<String, String> propertiesMap = testDesc.toProperties(); final TableSink<?> actualSink = TableFactoryService.find(StreamTableSinkFactory.class, propertiesMap) .createStreamTableSink(propertiesMap); assertEquals(expectedSink, actualSink); }
@Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return createElasticsearchUpsertTableSink( descriptorProperties.isValue(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND), descriptorProperties.getTableSchema(SCHEMA), getHosts(descriptorProperties), descriptorProperties.getString(CONNECTOR_INDEX), descriptorProperties.getString(CONNECTOR_DOCUMENT_TYPE), descriptorProperties.getOptionalString(CONNECTOR_KEY_DELIMITER).orElse(DEFAULT_KEY_DELIMITER), descriptorProperties.getOptionalString(CONNECTOR_KEY_NULL_LITERAL).orElse(DEFAULT_KEY_NULL_LITERAL), getSerializationSchema(properties), SUPPORTED_CONTENT_TYPE, getFailureHandler(descriptorProperties), getSinkOptions(descriptorProperties)); }
ElasticsearchUpsertTableSinkFactoryBase implements StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return createElasticsearchUpsertTableSink( descriptorProperties.isValue(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND), descriptorProperties.getTableSchema(SCHEMA), getHosts(descriptorProperties), descriptorProperties.getString(CONNECTOR_INDEX), descriptorProperties.getString(CONNECTOR_DOCUMENT_TYPE), descriptorProperties.getOptionalString(CONNECTOR_KEY_DELIMITER).orElse(DEFAULT_KEY_DELIMITER), descriptorProperties.getOptionalString(CONNECTOR_KEY_NULL_LITERAL).orElse(DEFAULT_KEY_NULL_LITERAL), getSerializationSchema(properties), SUPPORTED_CONTENT_TYPE, getFailureHandler(descriptorProperties), getSinkOptions(descriptorProperties)); } }
ElasticsearchUpsertTableSinkFactoryBase implements StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return createElasticsearchUpsertTableSink( descriptorProperties.isValue(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND), descriptorProperties.getTableSchema(SCHEMA), getHosts(descriptorProperties), descriptorProperties.getString(CONNECTOR_INDEX), descriptorProperties.getString(CONNECTOR_DOCUMENT_TYPE), descriptorProperties.getOptionalString(CONNECTOR_KEY_DELIMITER).orElse(DEFAULT_KEY_DELIMITER), descriptorProperties.getOptionalString(CONNECTOR_KEY_NULL_LITERAL).orElse(DEFAULT_KEY_NULL_LITERAL), getSerializationSchema(properties), SUPPORTED_CONTENT_TYPE, getFailureHandler(descriptorProperties), getSinkOptions(descriptorProperties)); }  }
ElasticsearchUpsertTableSinkFactoryBase implements StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return createElasticsearchUpsertTableSink( descriptorProperties.isValue(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND), descriptorProperties.getTableSchema(SCHEMA), getHosts(descriptorProperties), descriptorProperties.getString(CONNECTOR_INDEX), descriptorProperties.getString(CONNECTOR_DOCUMENT_TYPE), descriptorProperties.getOptionalString(CONNECTOR_KEY_DELIMITER).orElse(DEFAULT_KEY_DELIMITER), descriptorProperties.getOptionalString(CONNECTOR_KEY_NULL_LITERAL).orElse(DEFAULT_KEY_NULL_LITERAL), getSerializationSchema(properties), SUPPORTED_CONTENT_TYPE, getFailureHandler(descriptorProperties), getSinkOptions(descriptorProperties)); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties); }
ElasticsearchUpsertTableSinkFactoryBase implements StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return createElasticsearchUpsertTableSink( descriptorProperties.isValue(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND), descriptorProperties.getTableSchema(SCHEMA), getHosts(descriptorProperties), descriptorProperties.getString(CONNECTOR_INDEX), descriptorProperties.getString(CONNECTOR_DOCUMENT_TYPE), descriptorProperties.getOptionalString(CONNECTOR_KEY_DELIMITER).orElse(DEFAULT_KEY_DELIMITER), descriptorProperties.getOptionalString(CONNECTOR_KEY_NULL_LITERAL).orElse(DEFAULT_KEY_NULL_LITERAL), getSerializationSchema(properties), SUPPORTED_CONTENT_TYPE, getFailureHandler(descriptorProperties), getSinkOptions(descriptorProperties)); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties);  }