@Test public void shouldFailToConfigureWithInvalidSerializerClass() throws Exception { parameters.put(SERIALIZER, "java.lang.String"); try { Configurables.configure(fixture, new Context(parameters)); } catch (ClassCastException e) { } parameters.put(SERIALIZER, FakeConfigurable.class.getName()); try { Configurables.configure(fixture, new Context(parameters)); } catch (IllegalArgumentException e) { } }
@Override public void configure(Context context) { if (!isLocal) { String[] hostNames = null; if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) { hostNames = context.getString(HOSTNAMES).split(","); } Preconditions.checkState(hostNames != null && hostNames.length > 0, "Missing Param:" + HOSTNAMES); serverAddresses = new InetSocketTransportAddress[hostNames.length]; for (int i = 0; i < hostNames.length; i++) { String[] hostPort = hostNames[i].trim().split(":"); String host = hostPort[0].trim(); int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim()) : DEFAULT_PORT; serverAddresses[i] = new InetSocketTransportAddress(host, port); } Preconditions.checkState(serverAddresses != null && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES); } if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) { this.indexName = context.getString(INDEX_NAME); } if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) { this.indexType = context.getString(INDEX_TYPE); } if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) { this.clusterName = context.getString(CLUSTER_NAME); } if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) { this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE)); } if (StringUtils.isNotBlank(context.getString(TTL))) { this.ttlMs = parseTTL(context.getString(TTL)); Preconditions.checkState(ttlMs > 0, TTL + " must be greater than 0 or not set."); } String serializerClazz = "org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer"; if (StringUtils.isNotBlank(context.getString(SERIALIZER))) { serializerClazz = context.getString(SERIALIZER); } Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX)); try { @SuppressWarnings("unchecked") Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class .forName(serializerClazz); Configurable serializer = clazz.newInstance(); if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) { indexRequestFactory = (ElasticSearchIndexRequestBuilderFactory) serializer; } else if (serializer instanceof ElasticSearchEventSerializer){ indexRequestFactory = new EventSerializerIndexRequestBuilderFactory( (ElasticSearchEventSerializer) serializer); } else { throw new IllegalArgumentException( serializerClazz + " is neither an ElasticSearchEventSerializer" + " nor an ElasticSearchIndexRequestBuilderFactory."); } indexRequestFactory.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } Preconditions.checkState(StringUtils.isNotBlank(indexName), "Missing Param:" + INDEX_NAME); Preconditions.checkState(StringUtils.isNotBlank(indexType), "Missing Param:" + INDEX_TYPE); Preconditions.checkState(StringUtils.isNotBlank(clusterName), "Missing Param:" + CLUSTER_NAME); Preconditions.checkState(batchSize >= 1, BATCH_SIZE + " must be greater than 0"); }
ElasticSearchSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { if (!isLocal) { String[] hostNames = null; if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) { hostNames = context.getString(HOSTNAMES).split(","); } Preconditions.checkState(hostNames != null && hostNames.length > 0, "Missing Param:" + HOSTNAMES); serverAddresses = new InetSocketTransportAddress[hostNames.length]; for (int i = 0; i < hostNames.length; i++) { String[] hostPort = hostNames[i].trim().split(":"); String host = hostPort[0].trim(); int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim()) : DEFAULT_PORT; serverAddresses[i] = new InetSocketTransportAddress(host, port); } Preconditions.checkState(serverAddresses != null && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES); } if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) { this.indexName = context.getString(INDEX_NAME); } if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) { this.indexType = context.getString(INDEX_TYPE); } if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) { this.clusterName = context.getString(CLUSTER_NAME); } if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) { this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE)); } if (StringUtils.isNotBlank(context.getString(TTL))) { this.ttlMs = parseTTL(context.getString(TTL)); Preconditions.checkState(ttlMs > 0, TTL + " must be greater than 0 or not set."); } String serializerClazz = "org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer"; if (StringUtils.isNotBlank(context.getString(SERIALIZER))) { serializerClazz = context.getString(SERIALIZER); } Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX)); try { @SuppressWarnings("unchecked") Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class .forName(serializerClazz); Configurable serializer = clazz.newInstance(); if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) { indexRequestFactory = (ElasticSearchIndexRequestBuilderFactory) serializer; } else if (serializer instanceof ElasticSearchEventSerializer){ indexRequestFactory = new EventSerializerIndexRequestBuilderFactory( (ElasticSearchEventSerializer) serializer); } else { throw new IllegalArgumentException( serializerClazz + " is neither an ElasticSearchEventSerializer" + " nor an ElasticSearchIndexRequestBuilderFactory."); } indexRequestFactory.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } Preconditions.checkState(StringUtils.isNotBlank(indexName), "Missing Param:" + INDEX_NAME); Preconditions.checkState(StringUtils.isNotBlank(indexType), "Missing Param:" + INDEX_TYPE); Preconditions.checkState(StringUtils.isNotBlank(clusterName), "Missing Param:" + CLUSTER_NAME); Preconditions.checkState(batchSize >= 1, BATCH_SIZE + " must be greater than 0"); } }
ElasticSearchSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { if (!isLocal) { String[] hostNames = null; if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) { hostNames = context.getString(HOSTNAMES).split(","); } Preconditions.checkState(hostNames != null && hostNames.length > 0, "Missing Param:" + HOSTNAMES); serverAddresses = new InetSocketTransportAddress[hostNames.length]; for (int i = 0; i < hostNames.length; i++) { String[] hostPort = hostNames[i].trim().split(":"); String host = hostPort[0].trim(); int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim()) : DEFAULT_PORT; serverAddresses[i] = new InetSocketTransportAddress(host, port); } Preconditions.checkState(serverAddresses != null && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES); } if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) { this.indexName = context.getString(INDEX_NAME); } if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) { this.indexType = context.getString(INDEX_TYPE); } if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) { this.clusterName = context.getString(CLUSTER_NAME); } if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) { this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE)); } if (StringUtils.isNotBlank(context.getString(TTL))) { this.ttlMs = parseTTL(context.getString(TTL)); Preconditions.checkState(ttlMs > 0, TTL + " must be greater than 0 or not set."); } String serializerClazz = "org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer"; if (StringUtils.isNotBlank(context.getString(SERIALIZER))) { serializerClazz = context.getString(SERIALIZER); } Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX)); try { @SuppressWarnings("unchecked") Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class .forName(serializerClazz); Configurable serializer = clazz.newInstance(); if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) { indexRequestFactory = (ElasticSearchIndexRequestBuilderFactory) serializer; } else if (serializer instanceof ElasticSearchEventSerializer){ indexRequestFactory = new EventSerializerIndexRequestBuilderFactory( (ElasticSearchEventSerializer) serializer); } else { throw new IllegalArgumentException( serializerClazz + " is neither an ElasticSearchEventSerializer" + " nor an ElasticSearchIndexRequestBuilderFactory."); } indexRequestFactory.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } Preconditions.checkState(StringUtils.isNotBlank(indexName), "Missing Param:" + INDEX_NAME); Preconditions.checkState(StringUtils.isNotBlank(indexType), "Missing Param:" + INDEX_TYPE); Preconditions.checkState(StringUtils.isNotBlank(clusterName), "Missing Param:" + CLUSTER_NAME); Preconditions.checkState(batchSize >= 1, BATCH_SIZE + " must be greater than 0"); } ElasticSearchSink(); @VisibleForTesting  ElasticSearchSink(boolean isLocal); }
ElasticSearchSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { if (!isLocal) { String[] hostNames = null; if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) { hostNames = context.getString(HOSTNAMES).split(","); } Preconditions.checkState(hostNames != null && hostNames.length > 0, "Missing Param:" + HOSTNAMES); serverAddresses = new InetSocketTransportAddress[hostNames.length]; for (int i = 0; i < hostNames.length; i++) { String[] hostPort = hostNames[i].trim().split(":"); String host = hostPort[0].trim(); int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim()) : DEFAULT_PORT; serverAddresses[i] = new InetSocketTransportAddress(host, port); } Preconditions.checkState(serverAddresses != null && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES); } if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) { this.indexName = context.getString(INDEX_NAME); } if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) { this.indexType = context.getString(INDEX_TYPE); } if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) { this.clusterName = context.getString(CLUSTER_NAME); } if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) { this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE)); } if (StringUtils.isNotBlank(context.getString(TTL))) { this.ttlMs = parseTTL(context.getString(TTL)); Preconditions.checkState(ttlMs > 0, TTL + " must be greater than 0 or not set."); } String serializerClazz = "org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer"; if (StringUtils.isNotBlank(context.getString(SERIALIZER))) { serializerClazz = context.getString(SERIALIZER); } Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX)); try { @SuppressWarnings("unchecked") Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class .forName(serializerClazz); Configurable serializer = clazz.newInstance(); if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) { indexRequestFactory = (ElasticSearchIndexRequestBuilderFactory) serializer; } else if (serializer instanceof ElasticSearchEventSerializer){ indexRequestFactory = new EventSerializerIndexRequestBuilderFactory( (ElasticSearchEventSerializer) serializer); } else { throw new IllegalArgumentException( serializerClazz + " is neither an ElasticSearchEventSerializer" + " nor an ElasticSearchIndexRequestBuilderFactory."); } indexRequestFactory.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } Preconditions.checkState(StringUtils.isNotBlank(indexName), "Missing Param:" + INDEX_NAME); Preconditions.checkState(StringUtils.isNotBlank(indexType), "Missing Param:" + INDEX_TYPE); Preconditions.checkState(StringUtils.isNotBlank(clusterName), "Missing Param:" + CLUSTER_NAME); Preconditions.checkState(batchSize >= 1, BATCH_SIZE + " must be greater than 0"); } ElasticSearchSink(); @VisibleForTesting  ElasticSearchSink(boolean isLocal); @Override Status process(); @Override void configure(Context context); @Override void start(); @Override void stop(); }
ElasticSearchSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { if (!isLocal) { String[] hostNames = null; if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) { hostNames = context.getString(HOSTNAMES).split(","); } Preconditions.checkState(hostNames != null && hostNames.length > 0, "Missing Param:" + HOSTNAMES); serverAddresses = new InetSocketTransportAddress[hostNames.length]; for (int i = 0; i < hostNames.length; i++) { String[] hostPort = hostNames[i].trim().split(":"); String host = hostPort[0].trim(); int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim()) : DEFAULT_PORT; serverAddresses[i] = new InetSocketTransportAddress(host, port); } Preconditions.checkState(serverAddresses != null && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES); } if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) { this.indexName = context.getString(INDEX_NAME); } if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) { this.indexType = context.getString(INDEX_TYPE); } if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) { this.clusterName = context.getString(CLUSTER_NAME); } if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) { this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE)); } if (StringUtils.isNotBlank(context.getString(TTL))) { this.ttlMs = parseTTL(context.getString(TTL)); Preconditions.checkState(ttlMs > 0, TTL + " must be greater than 0 or not set."); } String serializerClazz = "org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer"; if (StringUtils.isNotBlank(context.getString(SERIALIZER))) { serializerClazz = context.getString(SERIALIZER); } Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX)); try { @SuppressWarnings("unchecked") Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class .forName(serializerClazz); Configurable serializer = clazz.newInstance(); if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) { indexRequestFactory = (ElasticSearchIndexRequestBuilderFactory) serializer; } else if (serializer instanceof ElasticSearchEventSerializer){ indexRequestFactory = new EventSerializerIndexRequestBuilderFactory( (ElasticSearchEventSerializer) serializer); } else { throw new IllegalArgumentException( serializerClazz + " is neither an ElasticSearchEventSerializer" + " nor an ElasticSearchIndexRequestBuilderFactory."); } indexRequestFactory.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } Preconditions.checkState(StringUtils.isNotBlank(indexName), "Missing Param:" + INDEX_NAME); Preconditions.checkState(StringUtils.isNotBlank(indexType), "Missing Param:" + INDEX_TYPE); Preconditions.checkState(StringUtils.isNotBlank(clusterName), "Missing Param:" + CLUSTER_NAME); Preconditions.checkState(batchSize >= 1, BATCH_SIZE + " must be greater than 0"); } ElasticSearchSink(); @VisibleForTesting  ElasticSearchSink(boolean isLocal); @Override Status process(); @Override void configure(Context context); @Override void start(); @Override void stop();  }