@Override
    public void process(Tuple input) {
        String key = filterMapper.getKeyFromTuple(input);

        boolean found;
        JedisCommandsContainer jedisCommand = null;
        try {
            jedisCommand = getInstance();

            switch (dataType) {
                case STRING:
                    found = jedisCommand.exists(key);
                    break;

                case SET:
                    found = jedisCommand.sismember(additionalKey, key);
                    break;

                case HASH:
                    found = jedisCommand.hexists(additionalKey, key);
                    break;

                case SORTED_SET:
                    found = jedisCommand.zrank(additionalKey, key) != null;
                    break;

                case HYPER_LOG_LOG:
                    found = jedisCommand.pfcount(key) > 0;
                    break;

                case GEO:
                    List<GeoCoordinate> geopos = jedisCommand.geopos(additionalKey, key);
                    if (geopos == null || geopos.isEmpty()) {
                        found = false;
                    } else {
                        // If any entry is NOT null, then we have a match.
                        found = geopos.stream()
                            .anyMatch(Objects::nonNull);
                    }
                    break;

                default:
                    throw new IllegalArgumentException("Cannot process such data type: " + dataType);
            }

            if (found) {
                collector.emit(input, input.getValues());
            }

            collector.ack(input);
        } catch (Exception e) {
            this.collector.reportError(e);
            this.collector.fail(input);
        }
    }
@Test
    void smokeTest_exists_keyFound() {
        // Define input key
        final String inputKey = "ThisIsMyKey";

        // Ensure key does exist in redis
        jedisHelper.set(inputKey, "some-value");
        assertTrue(jedisHelper.exists(inputKey), "Sanity check key exists.");

        // Create an input tuple
        final Map<String, Object> values = new HashMap<>();
        values.put("key", inputKey);
        values.put("value", "ThisIsMyValue");
        final Tuple tuple = new StubTuple(values);

        final JedisPoolConfig config = configBuilder.build();
        final TestMapper mapper = new TestMapper(STRING);

        final RedisFilterBolt bolt = new RedisFilterBolt(config, mapper);
        bolt.prepare(new HashMap<>(), topologyContext, new OutputCollector(outputCollector));
        bolt.process(tuple);

        // Verify Tuple passed through the bolt
        verifyTuplePassed(tuple);
    }