out_forward: add warning for zstd as experimental (#4896)

**Which issue(s) this PR fixes**: 
* Related to #4758

**What this PR does / why we need it**: 
When using `compress zstd` for `out_forward`, output the following
warning log.

```
zstd compression feature is an experimental new feature supported since v1.19.0.
Please make sure that the destination server also supports this feature before using it.
in_forward plugin for Fluentd supports it since v1.19.0.
```

In #4758, it was agreed that we would refrain from updating [the forward
protocol
v1.5](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1.5)
for the time being and instead output a log message indicating that this
feature is experimental.

> for now, we will make zstd available as an exprimenatal feature on the
Fluentd side, and note in the documentation that it is an experimental
feature that cannot be used unless the server supports it.

> Just displaying as a warning log to tell: "This is experimental
feature and not standardized yet" or similar message should be enough.

* #4758 (comment)
* #4758 (comment)
* #4758 (comment)

**Docs Changes**:
Not needed. (It should be done as #4657)

**Release Note**: 
Not needed. (A note for #4657 would be enough.)

Signed-off-by: Daijiro Fukuda <fukuda@clear-code.com>

Author: daipom

lib/fluent/plugin/out_forward.rb

@@ -271,9 +271,15 @@ def configure(conf)
       end
 
       raise Fluent::ConfigError, "ack_response_timeout must be a positive integer" if @ack_response_timeout < 1
+
+      if @compress == :zstd
+        log.warn "zstd compression feature is an experimental new feature supported since v1.19.0." +
+                 " Please make sure that the destination server also supports this feature before using it." +
+                 " in_forward plugin for Fluentd supports it since v1.19.0."
+      end
+
       @healthy_nodes_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "healthy_nodes_count", help_text: "Number of count healthy nodes", prefer_gauge: true)
       @registered_nodes_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "registered_nodes_count", help_text: "Number of count registered nodes", prefer_gauge: true)
-
     end
 
     def multi_workers_ready?

test/plugin/test_out_forward.rb

@@ -178,6 +178,60 @@ def try_write(chunk)
     assert{ logs.any?{|log| log.include?(expected_log) && log.include?(expected_detail) } }
   end
 
+  sub_test_case 'configure compress' do
+    data('default', ['', :text])
+    data('gzip', ['compress gzip', :gzip])
+    data('zstd', ['compress zstd', :zstd])
+    test 'should be applied' do |(option, expected)|
+      @d = d = create_driver(config + option)
+      node = d.instance.nodes.first
+
+      assert_equal(
+        [expected, expected],
+        [d.instance.compress, node.instance_variable_get(:@compress)]
+      )
+    end
+
+    data('default' => '')
+    data('gzip' => 'compress gzip')
+    data('zstd' => 'compress zstd')
+    test 'should log as experimental only for zstd' do |option|
+      @d = d = create_driver(config + option)
+
+      log_message = "zstd compression feature is an experimental new feature"
+      assert do
+        if d.instance.compress == :zstd
+          d.logs.any? { |log| log.include?(log_message) }
+        else
+          d.logs.none? { |log| log.include?(log_message) }
+        end
+      end
+    end
+
+    # TODO add tests that we cannot configure the different compress type between owner and buffer except for :text
+    data('gzip', ['compress gzip', :text, :gzip])
+    data('zstd', ['compress zstd', :text, :zstd])
+    test 'can configure buffer compress separately when owner uses :text' do |(buffer_option, expected_owner_compress, expected_buffer_compress)|
+      @d = d = create_driver(config + %[
+         <buffer>
+           type memory
+           #{buffer_option}
+         </buffer>
+       ])
+      node = d.instance.nodes.first
+
+      assert_equal(
+        [expected_owner_compress, expected_owner_compress, expected_buffer_compress],
+        [d.instance.compress, node.instance_variable_get(:@compress), d.instance.buffer.compress],
+      )
+
+      log_message = "buffer is compressed.  If you also want to save the bandwidth of a network, Add `compress` configuration in <match>"
+      assert do
+        d.logs.any? { |log| log.include?(log_message) }
+      end
+    end
+  end
+
   data('CA cert'     => 'tls_ca_cert_path',
        'non CA cert' => 'tls_cert_path')
   test 'configure tls_cert_path/tls_ca_cert_path' do |param|
@@ -326,66 +380,6 @@ def try_write(chunk)
     assert_equal 1234, d.instance.discovery_manager.services[0].port
   end
 
-  test 'compress_default_value' do
-    @d = d = create_driver
-    assert_equal :text, d.instance.compress
-
-    node = d.instance.nodes.first
-    assert_equal :text, node.instance_variable_get(:@compress)
-  end
-
-  test 'set_compress_is_gzip' do
-    @d = d = create_driver(config + %[compress gzip])
-    assert_equal :gzip, d.instance.compress
-    assert_equal :gzip, d.instance.buffer.compress
-
-    node = d.instance.nodes.first
-    assert_equal :gzip, node.instance_variable_get(:@compress)
-  end
-
-  test 'set_compress_is_zstd' do
-    @d = d = create_driver(config + %[compress zstd])
-    assert_equal :zstd, d.instance.compress
-    assert_equal :zstd, d.instance.buffer.compress
-
-    node = d.instance.nodes.first
-    assert_equal :zstd, node.instance_variable_get(:@compress)
-  end
-
-  test 'set_compress_is_gzip_in_buffer_section' do
-    mock = flexmock($log)
-    mock.should_receive(:log).with("buffer is compressed.  If you also want to save the bandwidth of a network, Add `compress` configuration in <match>")
-
-    @d = d = create_driver(config + %[
-       <buffer>
-         type memory
-         compress gzip
-       </buffer>
-     ])
-    assert_equal :text, d.instance.compress
-    assert_equal :gzip, d.instance.buffer.compress
-
-    node = d.instance.nodes.first
-    assert_equal :text, node.instance_variable_get(:@compress)
-  end
-
-  test 'set_compress_is_zstd_in_buffer_section' do
-    mock = flexmock($log)
-    mock.should_receive(:log).with("buffer is compressed.  If you also want to save the bandwidth of a network, Add `compress` configuration in <match>")
-
-    @d = d = create_driver(config + %[
-       <buffer>
-         type memory
-         compress zstd
-       </buffer>
-     ])
-    assert_equal :text, d.instance.compress
-    assert_equal :zstd, d.instance.buffer.compress
-
-    node = d.instance.nodes.first
-    assert_equal :text, node.instance_variable_get(:@compress)
-  end
-
   test 'phi_failure_detector disabled' do
     @d = d = create_driver(config + %[phi_failure_detector false \n phi_threshold 0])
     node = d.instance.nodes.first