@Test public void thatParseWorksOnIgnoreCase() throws EventDeliveryException { final CassandraSink sink = new CassandraSink(); final Channel channel = mock(Channel.class); final Transaction tx = mock(Transaction.class); final Session session = mock(Session.class); final ConsistencyLevel consistencyLevel = ConsistencyLevel.QUORUM; String bodyColumn = null; TableMetadata tableMetadata = mock(TableMetadata.class); final CassandraTable tableWithoutIgnoreCase = new CassandraTable(session, tableMetadata, consistencyLevel, bodyColumn); final Context ctx = new Context(); ctx.put("tables", "keyspace.table"); sink.configure(ctx); sink.tables = Collections.singletonList(tableWithoutIgnoreCase); sink.setChannel(channel); when(channel.getTransaction()).thenReturn(tx); mockTableMetadataWithIdAndNameColumns(tableMetadata); final Event event = EventBuilder.withBody(new byte[0], ImmutableMap.of("ID", "1", "NAME", "text")); final Map<String, Object> parsedResult = tableWithoutIgnoreCase.parse(event); assertThat(parsedResult).isEmpty(); final CassandraTable tableWitIgnoreCase = new CassandraTable(session, tableMetadata, consistencyLevel, bodyColumn, true); ctx.put("ignoreCase", "true"); final Map<String, Object> parsedResultIgnoreCase = tableWitIgnoreCase.parse(event); assertThat(parsedResultIgnoreCase).isNotEmpty(); assertThat(parsedResultIgnoreCase.get("id")).isNotNull(); }
@Override public void configure(Context context) { contactPoints = new ArrayList<InetSocketAddress>(); final String hosts = context.getString(CONF_HOSTS, DEFAULT_HOST); for (final String host : Splitter.on(',').split(hosts)) { try { final HostAndPort hostAndPort = HostAndPort.fromString(host).withDefaultPort(DEFAULT_PORT); contactPoints.add(new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort())); } catch (IllegalArgumentException ex) { throw new ConfigurationException("Could not parse host: " + host, ex); } } this.username = context.getString(CONF_USERNAME); this.password = context.getString(CONF_PASSWORD); this.consistency = context.getString(CONF_CONSISTENCY_LEVEL, DEFAULT_CONSISTENCY_LEVEL); this.bodyColumn = context.getString(CONF_BODY_COLUMN, DEFAULT_BODY_COLUMN); this.ignoreCase = context.getBoolean(CONF_IGNORE_CASE, DEFAULT_IGNORE_CASE_OPTION); final String tablesString = StringUtils.trimToNull(context.getString(CONF_TABLES)); if (tablesString == null) { throw new ConfigurationException(String.format("%s is mandatory", CONF_TABLES)); } this.tableStrings = Arrays.asList(tablesString.split(",")); final String cqlFile = StringUtils.trimToNull(context.getString(CONF_CQL_FILE)); if (cqlFile != null) { try { this.initCql = IOUtils.toString(new FileInputStream(cqlFile)); } catch (IOException ex) { throw new ConfigurationException("Cannot read CQL file: " + cqlFile, ex); } } this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.sinkCounter = new SinkCounter(this.getName()); }
CassandraSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { contactPoints = new ArrayList<InetSocketAddress>(); final String hosts = context.getString(CONF_HOSTS, DEFAULT_HOST); for (final String host : Splitter.on(',').split(hosts)) { try { final HostAndPort hostAndPort = HostAndPort.fromString(host).withDefaultPort(DEFAULT_PORT); contactPoints.add(new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort())); } catch (IllegalArgumentException ex) { throw new ConfigurationException("Could not parse host: " + host, ex); } } this.username = context.getString(CONF_USERNAME); this.password = context.getString(CONF_PASSWORD); this.consistency = context.getString(CONF_CONSISTENCY_LEVEL, DEFAULT_CONSISTENCY_LEVEL); this.bodyColumn = context.getString(CONF_BODY_COLUMN, DEFAULT_BODY_COLUMN); this.ignoreCase = context.getBoolean(CONF_IGNORE_CASE, DEFAULT_IGNORE_CASE_OPTION); final String tablesString = StringUtils.trimToNull(context.getString(CONF_TABLES)); if (tablesString == null) { throw new ConfigurationException(String.format("%s is mandatory", CONF_TABLES)); } this.tableStrings = Arrays.asList(tablesString.split(",")); final String cqlFile = StringUtils.trimToNull(context.getString(CONF_CQL_FILE)); if (cqlFile != null) { try { this.initCql = IOUtils.toString(new FileInputStream(cqlFile)); } catch (IOException ex) { throw new ConfigurationException("Cannot read CQL file: " + cqlFile, ex); } } this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.sinkCounter = new SinkCounter(this.getName()); } }
CassandraSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { contactPoints = new ArrayList<InetSocketAddress>(); final String hosts = context.getString(CONF_HOSTS, DEFAULT_HOST); for (final String host : Splitter.on(',').split(hosts)) { try { final HostAndPort hostAndPort = HostAndPort.fromString(host).withDefaultPort(DEFAULT_PORT); contactPoints.add(new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort())); } catch (IllegalArgumentException ex) { throw new ConfigurationException("Could not parse host: " + host, ex); } } this.username = context.getString(CONF_USERNAME); this.password = context.getString(CONF_PASSWORD); this.consistency = context.getString(CONF_CONSISTENCY_LEVEL, DEFAULT_CONSISTENCY_LEVEL); this.bodyColumn = context.getString(CONF_BODY_COLUMN, DEFAULT_BODY_COLUMN); this.ignoreCase = context.getBoolean(CONF_IGNORE_CASE, DEFAULT_IGNORE_CASE_OPTION); final String tablesString = StringUtils.trimToNull(context.getString(CONF_TABLES)); if (tablesString == null) { throw new ConfigurationException(String.format("%s is mandatory", CONF_TABLES)); } this.tableStrings = Arrays.asList(tablesString.split(",")); final String cqlFile = StringUtils.trimToNull(context.getString(CONF_CQL_FILE)); if (cqlFile != null) { try { this.initCql = IOUtils.toString(new FileInputStream(cqlFile)); } catch (IOException ex) { throw new ConfigurationException("Cannot read CQL file: " + cqlFile, ex); } } this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.sinkCounter = new SinkCounter(this.getName()); } CassandraSink(); }
CassandraSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { contactPoints = new ArrayList<InetSocketAddress>(); final String hosts = context.getString(CONF_HOSTS, DEFAULT_HOST); for (final String host : Splitter.on(',').split(hosts)) { try { final HostAndPort hostAndPort = HostAndPort.fromString(host).withDefaultPort(DEFAULT_PORT); contactPoints.add(new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort())); } catch (IllegalArgumentException ex) { throw new ConfigurationException("Could not parse host: " + host, ex); } } this.username = context.getString(CONF_USERNAME); this.password = context.getString(CONF_PASSWORD); this.consistency = context.getString(CONF_CONSISTENCY_LEVEL, DEFAULT_CONSISTENCY_LEVEL); this.bodyColumn = context.getString(CONF_BODY_COLUMN, DEFAULT_BODY_COLUMN); this.ignoreCase = context.getBoolean(CONF_IGNORE_CASE, DEFAULT_IGNORE_CASE_OPTION); final String tablesString = StringUtils.trimToNull(context.getString(CONF_TABLES)); if (tablesString == null) { throw new ConfigurationException(String.format("%s is mandatory", CONF_TABLES)); } this.tableStrings = Arrays.asList(tablesString.split(",")); final String cqlFile = StringUtils.trimToNull(context.getString(CONF_CQL_FILE)); if (cqlFile != null) { try { this.initCql = IOUtils.toString(new FileInputStream(cqlFile)); } catch (IOException ex) { throw new ConfigurationException("Cannot read CQL file: " + cqlFile, ex); } } this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.sinkCounter = new SinkCounter(this.getName()); } CassandraSink(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process(); }
CassandraSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { contactPoints = new ArrayList<InetSocketAddress>(); final String hosts = context.getString(CONF_HOSTS, DEFAULT_HOST); for (final String host : Splitter.on(',').split(hosts)) { try { final HostAndPort hostAndPort = HostAndPort.fromString(host).withDefaultPort(DEFAULT_PORT); contactPoints.add(new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort())); } catch (IllegalArgumentException ex) { throw new ConfigurationException("Could not parse host: " + host, ex); } } this.username = context.getString(CONF_USERNAME); this.password = context.getString(CONF_PASSWORD); this.consistency = context.getString(CONF_CONSISTENCY_LEVEL, DEFAULT_CONSISTENCY_LEVEL); this.bodyColumn = context.getString(CONF_BODY_COLUMN, DEFAULT_BODY_COLUMN); this.ignoreCase = context.getBoolean(CONF_IGNORE_CASE, DEFAULT_IGNORE_CASE_OPTION); final String tablesString = StringUtils.trimToNull(context.getString(CONF_TABLES)); if (tablesString == null) { throw new ConfigurationException(String.format("%s is mandatory", CONF_TABLES)); } this.tableStrings = Arrays.asList(tablesString.split(",")); final String cqlFile = StringUtils.trimToNull(context.getString(CONF_CQL_FILE)); if (cqlFile != null) { try { this.initCql = IOUtils.toString(new FileInputStream(cqlFile)); } catch (IOException ex) { throw new ConfigurationException("Cannot read CQL file: " + cqlFile, ex); } } this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.sinkCounter = new SinkCounter(this.getName()); } CassandraSink(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process();  }