@Test public void testRecommendedValues() { List<Object> expectedStorageClasses = Arrays.<Object>asList(S3Storage.class); List<Object> expectedFormatClasses = Arrays.<Object>asList( AvroFormat.class, JsonFormat.class, ByteArrayFormat.class, ParquetFormat.class ); List<Object> expectedPartitionerClasses = Arrays.<Object>asList( DefaultPartitioner.class, HourlyPartitioner.class, DailyPartitioner.class, TimeBasedPartitioner.class, FieldPartitioner.class ); List<ConfigValue> values = S3SinkConnectorConfig.getConfig().validate(properties); for (ConfigValue val : values) { if (val.value() instanceof Class) { switch (val.name()) { case StorageCommonConfig.STORAGE_CLASS_CONFIG: assertEquals(expectedStorageClasses, val.recommendedValues()); break; case S3SinkConnectorConfig.FORMAT_CLASS_CONFIG: assertEquals(expectedFormatClasses, val.recommendedValues()); break; case PartitionerConfig.PARTITIONER_CLASS_CONFIG: assertEquals(expectedPartitionerClasses, val.recommendedValues()); break; } } } }
public static ConfigDef getConfig() { Set<String> skip = new HashSet<>(); skip.add(StorageSinkConnectorConfig.SHUTDOWN_TIMEOUT_CONFIG); ConfigDef visible = new ConfigDef(); addAllConfigKeys(visible, newConfigDef(), skip); addAllConfigKeys(visible, StorageCommonConfig.newConfigDef(STORAGE_CLASS_RECOMMENDER), skip); addAllConfigKeys(visible, PartitionerConfig.newConfigDef(PARTITIONER_CLASS_RECOMMENDER), skip); return visible; }
S3SinkConnectorConfig extends StorageSinkConnectorConfig { public static ConfigDef getConfig() { Set<String> skip = new HashSet<>(); skip.add(StorageSinkConnectorConfig.SHUTDOWN_TIMEOUT_CONFIG); ConfigDef visible = new ConfigDef(); addAllConfigKeys(visible, newConfigDef(), skip); addAllConfigKeys(visible, StorageCommonConfig.newConfigDef(STORAGE_CLASS_RECOMMENDER), skip); addAllConfigKeys(visible, PartitionerConfig.newConfigDef(PARTITIONER_CLASS_RECOMMENDER), skip); return visible; } }
S3SinkConnectorConfig extends StorageSinkConnectorConfig { public static ConfigDef getConfig() { Set<String> skip = new HashSet<>(); skip.add(StorageSinkConnectorConfig.SHUTDOWN_TIMEOUT_CONFIG); ConfigDef visible = new ConfigDef(); addAllConfigKeys(visible, newConfigDef(), skip); addAllConfigKeys(visible, StorageCommonConfig.newConfigDef(STORAGE_CLASS_RECOMMENDER), skip); addAllConfigKeys(visible, PartitionerConfig.newConfigDef(PARTITIONER_CLASS_RECOMMENDER), skip); return visible; } S3SinkConnectorConfig(Map<String, String> props); protected  S3SinkConnectorConfig(ConfigDef configDef, Map<String, String> props); }
S3SinkConnectorConfig extends StorageSinkConnectorConfig { public static ConfigDef getConfig() { Set<String> skip = new HashSet<>(); skip.add(StorageSinkConnectorConfig.SHUTDOWN_TIMEOUT_CONFIG); ConfigDef visible = new ConfigDef(); addAllConfigKeys(visible, newConfigDef(), skip); addAllConfigKeys(visible, StorageCommonConfig.newConfigDef(STORAGE_CLASS_RECOMMENDER), skip); addAllConfigKeys(visible, PartitionerConfig.newConfigDef(PARTITIONER_CLASS_RECOMMENDER), skip); return visible; } S3SinkConnectorConfig(Map<String, String> props); protected  S3SinkConnectorConfig(ConfigDef configDef, Map<String, String> props); static ConfigDef newConfigDef(); String getBucketName(); String getSsea(); String getSseCustomerKey(); String getSseKmsKeyId(); boolean useExpectContinue(); CannedAccessControlList getCannedAcl(); int getPartSize(); @SuppressWarnings("unchecked") AWSCredentialsProvider getCredentialsProvider(); CompressionType getCompressionType(); int getCompressionLevel(); CompressionCodecName parquetCompressionCodecName(); int getS3PartRetries(); String getByteArrayExtension(); String getFormatByteArrayLineSeparator(); String getName(); @Override Object get(String key); Map<String, ?> plainValues(); static ConfigDef getConfig(); String nullValueBehavior(); static void main(String[] args); }
S3SinkConnectorConfig extends StorageSinkConnectorConfig { public static ConfigDef getConfig() { Set<String> skip = new HashSet<>(); skip.add(StorageSinkConnectorConfig.SHUTDOWN_TIMEOUT_CONFIG); ConfigDef visible = new ConfigDef(); addAllConfigKeys(visible, newConfigDef(), skip); addAllConfigKeys(visible, StorageCommonConfig.newConfigDef(STORAGE_CLASS_RECOMMENDER), skip); addAllConfigKeys(visible, PartitionerConfig.newConfigDef(PARTITIONER_CLASS_RECOMMENDER), skip); return visible; } S3SinkConnectorConfig(Map<String, String> props); protected  S3SinkConnectorConfig(ConfigDef configDef, Map<String, String> props); static ConfigDef newConfigDef(); String getBucketName(); String getSsea(); String getSseCustomerKey(); String getSseKmsKeyId(); boolean useExpectContinue(); CannedAccessControlList getCannedAcl(); int getPartSize(); @SuppressWarnings("unchecked") AWSCredentialsProvider getCredentialsProvider(); CompressionType getCompressionType(); int getCompressionLevel(); CompressionCodecName parquetCompressionCodecName(); int getS3PartRetries(); String getByteArrayExtension(); String getFormatByteArrayLineSeparator(); String getName(); @Override Object get(String key); Map<String, ?> plainValues(); static ConfigDef getConfig(); String nullValueBehavior(); static void main(String[] args); static final String S3_BUCKET_CONFIG; static final String S3_OBJECT_TAGGING_CONFIG; static final boolean S3_OBJECT_TAGGING_DEFAULT; static final String SSEA_CONFIG; static final String SSEA_DEFAULT; static final String SSE_CUSTOMER_KEY; static final Password SSE_CUSTOMER_KEY_DEFAULT; static final String SSE_KMS_KEY_ID_CONFIG; static final String SSE_KMS_KEY_ID_DEFAULT; static final String PART_SIZE_CONFIG; static final int PART_SIZE_DEFAULT; static final String WAN_MODE_CONFIG; static final String CREDENTIALS_PROVIDER_CLASS_CONFIG; static final Class<? extends AWSCredentialsProvider> CREDENTIALS_PROVIDER_CLASS_DEFAULT; static final String CREDENTIALS_PROVIDER_CONFIG_PREFIX; static final String AWS_ACCESS_KEY_ID_CONFIG; static final String AWS_ACCESS_KEY_ID_DEFAULT; static final String AWS_SECRET_ACCESS_KEY_CONFIG; static final Password AWS_SECRET_ACCESS_KEY_DEFAULT; static final String REGION_CONFIG; static final String REGION_DEFAULT; static final String ACL_CANNED_CONFIG; static final String ACL_CANNED_DEFAULT; static final String COMPRESSION_TYPE_CONFIG; static final String COMPRESSION_TYPE_DEFAULT; static final String COMPRESSION_LEVEL_CONFIG; static final int COMPRESSION_LEVEL_DEFAULT; static final String S3_PART_RETRIES_CONFIG; static final int S3_PART_RETRIES_DEFAULT; static final String FORMAT_BYTEARRAY_EXTENSION_CONFIG; static final String FORMAT_BYTEARRAY_EXTENSION_DEFAULT; static final String FORMAT_BYTEARRAY_LINE_SEPARATOR_CONFIG; static final String FORMAT_BYTEARRAY_LINE_SEPARATOR_DEFAULT; static final String S3_PROXY_URL_CONFIG; static final String S3_PROXY_URL_DEFAULT; static final String S3_PROXY_USER_CONFIG; static final String S3_PROXY_USER_DEFAULT; static final String S3_PROXY_PASS_CONFIG; static final Password S3_PROXY_PASS_DEFAULT; static final String HEADERS_USE_EXPECT_CONTINUE_CONFIG; static final boolean HEADERS_USE_EXPECT_CONTINUE_DEFAULT; static final String BEHAVIOR_ON_NULL_VALUES_CONFIG; static final String BEHAVIOR_ON_NULL_VALUES_DEFAULT; static final int S3_RETRY_MAX_BACKOFF_TIME_MS; static final String S3_RETRY_BACKOFF_CONFIG; static final int S3_RETRY_BACKOFF_DEFAULT; static final String S3_PATH_STYLE_ACCESS_ENABLED_CONFIG; static final boolean S3_PATH_STYLE_ACCESS_ENABLED_DEFAULT; static final String STORE_KAFKA_KEYS_CONFIG; static final String STORE_KAFKA_HEADERS_CONFIG; static final String KEYS_FORMAT_CLASS_CONFIG; static final Class<? extends Format> KEYS_FORMAT_CLASS_DEFAULT; static final String HEADERS_FORMAT_CLASS_CONFIG; static final Class<? extends Format> HEADERS_FORMAT_CLASS_DEFAULT; }