@Test(expected = IllegalStateException.class) public void negativeBatchConfiguredTest() throws EventDeliveryException { Jedis jedis = mock(Jedis.class); JedisPool jedisPool = mock(JedisPool.class); MockJedisPoolFactory mockJedisPoolFactory = new MockJedisPoolFactory(jedisPool, jedis); Channel channel = mock(Channel.class); Transaction transactionMock = mock(Transaction.class); when(channel.getTransaction()).thenReturn(transactionMock); RedisSink redisSink = new RedisSink(mockJedisPoolFactory); redisSink.setChannel(channel); Context context = new Context(); context.put(RedisSinkConfigurationConstant.HOST, "localhost"); context.put(RedisSinkConfigurationConstant.BATCH_SIZE, "-10"); redisSink.configure(context); }
@Override public void configure(Context context) { logger.info("Configuring"); host = context.getString(RedisSinkConfigurationConstant.HOST); Preconditions.checkState(StringUtils.isNotBlank(host), "host cannot be empty, please specify in configuration file"); port = context.getInteger(RedisSinkConfigurationConstant.PORT, Protocol.DEFAULT_PORT); timeout = context.getInteger(RedisSinkConfigurationConstant.TIMEOUT, Protocol.DEFAULT_TIMEOUT); database = context.getInteger(RedisSinkConfigurationConstant.DATABASE, Protocol.DEFAULT_DATABASE); password = context.getString(RedisSinkConfigurationConstant.PASSWORD); redisKey = context.getString(RedisSinkConfigurationConstant.KEY, RedisSinkConfigurationConstant.DEFAULT_KEY) .getBytes(); batchSize = context.getInteger(RedisSinkConfigurationConstant.BATCH_SIZE, RedisSinkConfigurationConstant.DEFAULT_BATCH_SIZE); String serializerClassName = context.getString(RedisSinkConfigurationConstant.SERIALIZER, RedisSinkConfigurationConstant.DEFAULT_SERIALIZER_CLASS_NAME); Preconditions.checkState(batchSize > 0, RedisSinkConfigurationConstant.BATCH_SIZE + " parameter must be greater than 1"); try { @SuppressWarnings("unchecked") Class<? extends Serializer> clazz = (Class<? extends Serializer>) Class .forName(serializerClassName); serializer = clazz.newInstance(); Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(RedisSinkConfigurationConstant.SERIALIZER_PREFIX)); serializer.configure(serializerContext); } catch (ClassNotFoundException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (InstantiationException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (IllegalAccessException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } }
RedisSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { logger.info("Configuring"); host = context.getString(RedisSinkConfigurationConstant.HOST); Preconditions.checkState(StringUtils.isNotBlank(host), "host cannot be empty, please specify in configuration file"); port = context.getInteger(RedisSinkConfigurationConstant.PORT, Protocol.DEFAULT_PORT); timeout = context.getInteger(RedisSinkConfigurationConstant.TIMEOUT, Protocol.DEFAULT_TIMEOUT); database = context.getInteger(RedisSinkConfigurationConstant.DATABASE, Protocol.DEFAULT_DATABASE); password = context.getString(RedisSinkConfigurationConstant.PASSWORD); redisKey = context.getString(RedisSinkConfigurationConstant.KEY, RedisSinkConfigurationConstant.DEFAULT_KEY) .getBytes(); batchSize = context.getInteger(RedisSinkConfigurationConstant.BATCH_SIZE, RedisSinkConfigurationConstant.DEFAULT_BATCH_SIZE); String serializerClassName = context.getString(RedisSinkConfigurationConstant.SERIALIZER, RedisSinkConfigurationConstant.DEFAULT_SERIALIZER_CLASS_NAME); Preconditions.checkState(batchSize > 0, RedisSinkConfigurationConstant.BATCH_SIZE + " parameter must be greater than 1"); try { @SuppressWarnings("unchecked") Class<? extends Serializer> clazz = (Class<? extends Serializer>) Class .forName(serializerClassName); serializer = clazz.newInstance(); Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(RedisSinkConfigurationConstant.SERIALIZER_PREFIX)); serializer.configure(serializerContext); } catch (ClassNotFoundException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (InstantiationException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (IllegalAccessException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } } }
RedisSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { logger.info("Configuring"); host = context.getString(RedisSinkConfigurationConstant.HOST); Preconditions.checkState(StringUtils.isNotBlank(host), "host cannot be empty, please specify in configuration file"); port = context.getInteger(RedisSinkConfigurationConstant.PORT, Protocol.DEFAULT_PORT); timeout = context.getInteger(RedisSinkConfigurationConstant.TIMEOUT, Protocol.DEFAULT_TIMEOUT); database = context.getInteger(RedisSinkConfigurationConstant.DATABASE, Protocol.DEFAULT_DATABASE); password = context.getString(RedisSinkConfigurationConstant.PASSWORD); redisKey = context.getString(RedisSinkConfigurationConstant.KEY, RedisSinkConfigurationConstant.DEFAULT_KEY) .getBytes(); batchSize = context.getInteger(RedisSinkConfigurationConstant.BATCH_SIZE, RedisSinkConfigurationConstant.DEFAULT_BATCH_SIZE); String serializerClassName = context.getString(RedisSinkConfigurationConstant.SERIALIZER, RedisSinkConfigurationConstant.DEFAULT_SERIALIZER_CLASS_NAME); Preconditions.checkState(batchSize > 0, RedisSinkConfigurationConstant.BATCH_SIZE + " parameter must be greater than 1"); try { @SuppressWarnings("unchecked") Class<? extends Serializer> clazz = (Class<? extends Serializer>) Class .forName(serializerClassName); serializer = clazz.newInstance(); Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(RedisSinkConfigurationConstant.SERIALIZER_PREFIX)); serializer.configure(serializerContext); } catch (ClassNotFoundException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (InstantiationException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (IllegalAccessException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } } RedisSink(); @VisibleForTesting  RedisSink(JedisPoolFactory _jedisPoolFactory); }
RedisSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { logger.info("Configuring"); host = context.getString(RedisSinkConfigurationConstant.HOST); Preconditions.checkState(StringUtils.isNotBlank(host), "host cannot be empty, please specify in configuration file"); port = context.getInteger(RedisSinkConfigurationConstant.PORT, Protocol.DEFAULT_PORT); timeout = context.getInteger(RedisSinkConfigurationConstant.TIMEOUT, Protocol.DEFAULT_TIMEOUT); database = context.getInteger(RedisSinkConfigurationConstant.DATABASE, Protocol.DEFAULT_DATABASE); password = context.getString(RedisSinkConfigurationConstant.PASSWORD); redisKey = context.getString(RedisSinkConfigurationConstant.KEY, RedisSinkConfigurationConstant.DEFAULT_KEY) .getBytes(); batchSize = context.getInteger(RedisSinkConfigurationConstant.BATCH_SIZE, RedisSinkConfigurationConstant.DEFAULT_BATCH_SIZE); String serializerClassName = context.getString(RedisSinkConfigurationConstant.SERIALIZER, RedisSinkConfigurationConstant.DEFAULT_SERIALIZER_CLASS_NAME); Preconditions.checkState(batchSize > 0, RedisSinkConfigurationConstant.BATCH_SIZE + " parameter must be greater than 1"); try { @SuppressWarnings("unchecked") Class<? extends Serializer> clazz = (Class<? extends Serializer>) Class .forName(serializerClassName); serializer = clazz.newInstance(); Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(RedisSinkConfigurationConstant.SERIALIZER_PREFIX)); serializer.configure(serializerContext); } catch (ClassNotFoundException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (InstantiationException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (IllegalAccessException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } } RedisSink(); @VisibleForTesting  RedisSink(JedisPoolFactory _jedisPoolFactory); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process(); @Override void configure(Context context); }
RedisSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { logger.info("Configuring"); host = context.getString(RedisSinkConfigurationConstant.HOST); Preconditions.checkState(StringUtils.isNotBlank(host), "host cannot be empty, please specify in configuration file"); port = context.getInteger(RedisSinkConfigurationConstant.PORT, Protocol.DEFAULT_PORT); timeout = context.getInteger(RedisSinkConfigurationConstant.TIMEOUT, Protocol.DEFAULT_TIMEOUT); database = context.getInteger(RedisSinkConfigurationConstant.DATABASE, Protocol.DEFAULT_DATABASE); password = context.getString(RedisSinkConfigurationConstant.PASSWORD); redisKey = context.getString(RedisSinkConfigurationConstant.KEY, RedisSinkConfigurationConstant.DEFAULT_KEY) .getBytes(); batchSize = context.getInteger(RedisSinkConfigurationConstant.BATCH_SIZE, RedisSinkConfigurationConstant.DEFAULT_BATCH_SIZE); String serializerClassName = context.getString(RedisSinkConfigurationConstant.SERIALIZER, RedisSinkConfigurationConstant.DEFAULT_SERIALIZER_CLASS_NAME); Preconditions.checkState(batchSize > 0, RedisSinkConfigurationConstant.BATCH_SIZE + " parameter must be greater than 1"); try { @SuppressWarnings("unchecked") Class<? extends Serializer> clazz = (Class<? extends Serializer>) Class .forName(serializerClassName); serializer = clazz.newInstance(); Context serializerContext = new Context(); serializerContext.putAll(context.getSubProperties(RedisSinkConfigurationConstant.SERIALIZER_PREFIX)); serializer.configure(serializerContext); } catch (ClassNotFoundException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (InstantiationException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } catch (IllegalAccessException e) { logger.error("Could not instantiate event serializer", e); Throwables.propagate(e); } } RedisSink(); @VisibleForTesting  RedisSink(JedisPoolFactory _jedisPoolFactory); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process(); @Override void configure(Context context);  }