make queue_max_bytes a long again allowing queues bigger than 2GB (#18366)

The queue factory setting "queueMaxBytes" used to be a long [link](https://github.com/elastic/logstash/pull/18180/files#diff-a377851670e06cd113751aa73b726b446dfb721d03d6979b9195afff7facd9f5L60):

```ruby
long queueMaxBytes = RubyFixnum.num2long(args[6]);
```

But with the refactoring of #18180, it [became an Integer by mistake](https://github.com/elastic/logstash/pull/18180/files#diff-01eec670d2beabcd43041ff7fca8bc907eb227471dce886bb757b6414a655429R125):

```java
.queueMaxBytes(getSetting(context, settings, QUEUE_MAX_BYTES).toJava(Integer.class))
```
Which prevents queues being bigger than 2^31-1, causing crashes to anyone setting their PQ to 2GB or more:

```yaml
queue.max_bytes: 2147483647 # 2GB-1 OK
queue.max_bytes: 2147483648 # 2GB FAIL
```

This change makes queueMaxBytes a Long again, with a test confirming that queues larger than 2ˆ31-1 (max int) can be created.

Without this change the new test fails:

```
❯ bin/rspec logstash-core/spec/logstash/queue_factory_spec.rb
..F.

Failures:

  1) LogStash::QueueFactory when `queue.type` is `persisted` when queue.max_bytes is larger than Java int does not raise error
     Failure/Error: expect { queue = subject.create(settings) }.to_not raise_error

       expected no Exception, got #<RangeError: too big for int: 2147483648> with backtrace:
         # org/logstash/ackedqueue/QueueFactoryExt.java:97:in `create'
         # org/logstash/ackedqueue/QueueFactoryExt.java:88:in `create'
         # ./logstash-core/spec/logstash/queue_factory_spec.rb:85:in `block in <main>'
         # ./logstash-core/spec/logstash/queue_factory_spec.rb:85:in `block in <main>'
         # ./lib/bootstrap/rspec.rb:36:in `<main>'
     # ./logstash-core/spec/logstash/queue_factory_spec.rb:85:in `block in <main>'
     # ./lib/bootstrap/rspec.rb:36:in `<main>'

Finished in 0.03879 seconds (files took 0.03787 seconds to load)
4 examples, 1 failure

Failed examples:

rspec ./logstash-core/spec/logstash/queue_factory_spec.rb:84 # LogStash::QueueFactory when `queue.type` is `persisted` when queue.max_bytes is larger than Java int does not raise error
```

But with the change it passes:

```
❯ bin/rspec logstash-core/spec/logstash/queue_factory_spec.rb -fd

LogStash::QueueFactory
  when `queue.type` is `persisted`
    returns a `WrappedAckedQueue`
    per pipeline id subdirectory creation
      creates a queue directory based on the pipeline id
    when queue.max_bytes is larger than Java int
      does not raise error
  when `queue.type` is `memory`
    returns a `WrappedSynchronousQueue`

Finished in 0.03292 seconds (files took 0.04052 seconds to load)
4 examples, 0 failures
```

(cherry picked from commit 6c48e51)

Author: jsvd

logstash-core/spec/logstash/queue_factory_spec.rb

@@ -50,22 +50,23 @@
   subject { described_class }
 
   context "when `queue.type` is `persisted`" do
+    let(:queue_path) { ::File.join(settings.get("path.queue"), pipeline_id) }
+
     before do
       settings.set("queue.type", "persisted")
     end
 
+    after(:each) do
+      FileUtils.rm_rf(queue_path)
+    end
+
     it "returns a `WrappedAckedQueue`" do
       queue = subject.create(settings)
       expect(queue).to be_kind_of(LogStash::WrappedAckedQueue)
       queue.close
     end
 
     describe "per pipeline id subdirectory creation" do
-      let(:queue_path) { ::File.join(settings.get("path.queue"), pipeline_id) }
-
-      after :each do
-        FileUtils.rm_rf(queue_path)
-      end
 
       it "creates a queue directory based on the pipeline id" do
         expect(Dir.exist?(queue_path)).to be_falsey
@@ -74,6 +75,19 @@
         queue.close
       end
     end
+
+    context "when queue.max_bytes is larger than Java int" do
+      let(:large_queue_max_bytes) { 2**31 } # 2^31 bytes, bigger than 2^31-1 int limit
+      before(:each) do
+        settings.set("queue.max_bytes", large_queue_max_bytes)
+      end
+      it "does not raise error" do
+        queue = nil
+        expect { queue = subject.create(settings) }.to_not raise_error
+        expect(queue.queue.max_size_in_bytes).to eq(large_queue_max_bytes)
+        queue.close
+      end
+    end
   end
 
   context "when `queue.type` is `memory`" do

logstash-core/src/main/java/org/logstash/ackedqueue/QueueFactoryExt.java

@@ -160,7 +160,7 @@ private static Settings extractQueueSettings(final IRubyObject settings) {
                 .checkpointMaxWrites(getSetting(context, settings, QUEUE_CHECKPOINT_WRITES).toJava(Integer.class))
                 .checkpointMaxAcks(getSetting(context, settings, QUEUE_CHECKPOINT_ACKS).toJava(Integer.class))
                 .checkpointRetry(getSetting(context, settings, QUEUE_CHECKPOINT_RETRY).isTrue())
-                .queueMaxBytes(getSetting(context, settings, QUEUE_MAX_BYTES).toJava(Integer.class))
+                .queueMaxBytes(getSetting(context, settings, QUEUE_MAX_BYTES).toJava(Long.class))
                 .compressionCodecFactory(extractConfiguredCodec(settings))
                 .build();
     }