class RedisFeatureSink implements FeatureSink {

  @Override
  public PTransform<PCollection<FeatureRow>, WriteResult> writer() {
    int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS;

    if (getRedisClusterConfig() != null) {

      if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) {
        flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds();
      }

      RedisKeySerializer serializer =
          new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix());
      return new RedisCustomIO.Write(
              new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer)
          .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds))
          .withBatchSize(DEFAULT_BATCH_SIZE);

    } else if (getRedisConfig() != null) {
      if (getRedisConfig().getFlushFrequencySeconds() > 0) {
        flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds();
      }

      RedisKeySerializer serializer = new RedisKeyProtoSerializer();
      return new RedisCustomIO.Write(
              new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer)
          .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds))
          .withBatchSize(DEFAULT_BATCH_SIZE);
    } else {
      throw new RuntimeException(
          "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink");
    }
  }

}